orchestration_designate_prod_from_scratch.ipynb

{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### The \"genesis\" version: the db is in-development and no *consolidation* tables had been created; the resulting tables from civiActivityReport serve as the consolidation and stakjob source tables and there is no need to run the status and type stored procedures because there is no update to them\n",
    "This python file serves as an orchestration program to be run **after** *civiActivityReport.ipynb*.\n",
    "Components:\n",
    "1. rename (by way of copies) the two output tables from civiActivityReport.ipynb to consolidated_mem_type and consolidated_mem_status\n",
    "    This is done in order to preserve the original import table names, which should be deleted manually later\n",
    "2. call the stored procedure to create the stack_job table: *stored_procedure_create_stack_job.sql"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "import re\n",
    "import datetime\n",
    "import itertools\n",
    "import json\n",
    "from container_credentials import return_credentials\n",
    "import sqlalchemy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Connection to the 172.17.0.2 for user root created successfully.\n"
     ]
    }
   ],
   "source": [
    "# DEFINE THE DATABASE CREDENTIALS\n",
    "\n",
    "cred_dict = return_credentials()\n",
    "\n",
    "user = cred_dict['user'] \n",
    "password = cred_dict['pass'] \n",
    "host = cred_dict['host'] \n",
    "port = cred_dict['port'] \n",
    "database = cred_dict['database']\n",
    "\n",
    "''' legacy code hard coded the credentials\n",
    "user = 'root'\n",
    "password = 'baeldung'\n",
    "host = '172.17.0.2'\n",
    "port = 3306\n",
    "database = 'membership'\n",
    "'''\n",
    "\n",
    "def get_connection():\n",
    "\treturn sqlalchemy.create_engine(\n",
    "\t\turl=\"mysql+pymysql://{0}:{1}@{2}:{3}/{4}\".format(\n",
    "\t\t\tuser, password, host, port, database\n",
    "\t\t)\n",
    "\t)\n",
    "\n",
    "if __name__ == '__main__':\n",
    "\n",
    "\ttry:\n",
    "\t\n",
    "\t\t# GET THE CONNECTION OBJECT (ENGINE) FOR THE DATABASE\n",
    "\t\t# working w/engines: https://docs.sqlalchemy.org/en/20/core/engines_connections.html\n",
    "\t\tengine = get_connection() #engine should be created just once, and can manage several DBAPI connections\n",
    "\t\tprint(\n",
    "\t\t\tf\"Connection to the {host} for user {user} created successfully.\")\n",
    "\texcept Exception as ex:\n",
    "\t\tprint(\"Connection could not be made due to the following error: \\n\", ex)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "#make copies of the two output tables from the .ipynb <- change table name to a generic name to be consumed by the stored procedures\n",
    "#create copies of two tables at a time (in order) \"new_import\" suffix\n",
    "def copy_rename(type_table: str, status_table: str, type_table_replacement_name = None, status_table_replacement_name = None):\n",
    "    #a CORE approach\n",
    "    #type_table = 'mem_type_'\n",
    "    #status_table = 'mem_status_'\n",
    "    #want to limit the scope of the of our use of this object to a specific context, so we use Python's context manager \"with\"\n",
    "    with engine.connect() as conn: #interacting w/db through Connection class\n",
    "\n",
    "        if type_table_replacement_name is None:\n",
    "            conn.execute(sqlalchemy.text(f\"DROP TABLE IF EXISTS mem_type_new_import\"))\n",
    "            conn.execute(sqlalchemy.text(f\"CREATE TABLE mem_type_new_import LIKE {type_table}\"))\n",
    "            conn.execute(sqlalchemy.text(f\"INSERT INTO mem_type_new_import SELECT * FROM {type_table}\"))\n",
    "            conn.commit()\n",
    "        elif type_table_replacement_name is not None:\n",
    "            conn.execute(sqlalchemy.text(f\"DROP TABLE IF EXISTS {type_table_replacement_name}\"))\n",
    "            conn.execute(sqlalchemy.text(f\"CREATE TABLE {type_table_replacement_name} LIKE {type_table}\"))\n",
    "            conn.execute(sqlalchemy.text(f\"INSERT INTO {type_table_replacement_name} SELECT * FROM {type_table}\"))\n",
    "            conn.commit()\n",
    "        \n",
    "        if status_table_replacement_name is None:\n",
    "            conn.execute(sqlalchemy.text(f\"DROP TABLE IF EXISTS mem_status_new_import\"))\n",
    "            conn.execute(sqlalchemy.text(f\"CREATE TABLE mem_status_new_import LIKE {status_table}\"))\n",
    "            conn.execute(sqlalchemy.text(f\"INSERT INTO mem_status_new_import SELECT * FROM {status_table}\"))\n",
    "            conn.commit()\n",
    "        elif status_table_replacement_name is not None:\n",
    "            conn.execute(sqlalchemy.text(f\"DROP TABLE IF EXISTS {status_table_replacement_name}\"))\n",
    "            conn.execute(sqlalchemy.text(f\"CREATE TABLE {status_table_replacement_name} LIKE {status_table}\"))\n",
    "            conn.execute(sqlalchemy.text(f\"INSERT INTO {status_table_replacement_name} SELECT * FROM {status_table}\"))\n",
    "            conn.commit()\n",
    "\n",
    "        else:\n",
    "            raise SyntaxError(\"none of the four conditions for copy_rename were encountered\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "def qa_copy_rename(type_table: str, status_table: str, type_table_replacement_name, status_table_replacement_name):\n",
    "        #verify that table lengths/shapes are the same after the import\n",
    "        from sqlalchemy import text\n",
    "        with engine.connect() as conn:\n",
    "            type_orig = conn.execute(text(f\"SELECT COUNT(*) FROM {type_table}\")).scalar()\n",
    "            status_orig = conn.execute(text(f\"SELECT COUNT(*) FROM {status_table}\")).scalar()\n",
    "            type_post_import = conn.execute(text(f\"SELECT COUNT(*) FROM {type_table_replacement_name}\")).scalar()\n",
    "            status_post_import = conn.execute(text(f\"SELECT COUNT(*) FROM {status_table_replacement_name}\")).scalar()\n",
    "        \n",
    "        if type_orig == type_post_import and status_orig == status_post_import:\n",
    "            return 'tables are of expected type'\n",
    "        else:\n",
    "            raise RuntimeError(\"table sizes not as expected\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "#create variables for the import tables (will be used at least twice in the program)\n",
    "#expected that the 'type' and 'status' table names already follow the table_name_date convention\n",
    "mem_type_import = 'mem_type_1007_ts'\n",
    "mem_status_import = 'mem_status_1007_ts'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "# if I want to handle errors, the relative error category is \"programming-time error\"\n",
    "copy_rename(mem_type_import,mem_status_import,'consolidated_mem_type','consolidated_mem_status')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'tables are of expected type'"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#check copy_rename\n",
    "qa_copy_rename(mem_type_import,mem_status_import,'consolidated_mem_type','consolidated_mem_status')"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The only stored procedure necessary is the **stackjob_creations** procedure. Once that is run, the db will be equipped with consolidated_mem_type, consolidated_mem_status and stack_job2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['stackjob_creations', 'status_table_create', 'test_parameter', 'type_table_create']\n"
     ]
    }
   ],
   "source": [
    "with engine.connect() as conn:\n",
    "    result = conn.execute(sqlalchemy.text(\"show procedure status where definer LIKE '%root%'\"))\n",
    "    lista = [i[1] for i in result.all()]\n",
    "\n",
    "print(lista)    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "#TODO call the stack_job stored procedure <- the final input table to the active accounts study\n",
    "connection = engine.raw_connection()\n",
    "try:\n",
    "    cursor_obj = connection.cursor()\n",
    "    cursor_obj.callproc(\"stackjob_creations\")\n",
    "    cursor_obj.close()\n",
    "    connection.commit()\n",
    "finally:\n",
    "    connection.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [],
   "source": [
    "#TODO add some QA of the new stack_job2 table (from the stackjob_creation stored procedurei)\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "py12",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.2"
  },
  "orig_nbformat": 4
 },
 "nbformat": 4,
 "nbformat_minor": 2
}