TableETL.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "e773963d-2455-4fe4-bc5d-19b273a34a1b",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import numpy as np\n",
    "import sqlalchemy as sa\n",
    "from datetime import datetime, timedelta\n",
    "\n",
    "from collections import defaultdict\n",
    "from json import JSONEncoder,JSONDecoder\n",
    "import requests\n",
    "\n",
    "import threading\n",
    "from queue import Queue\n",
    "\n",
    "from time import sleep\n",
    "\n",
    "from simple_salesforce import Salesforce\n",
    "from os import environ, sys\n",
    "from dotenv import load_dotenv\n",
    "_ = load_dotenv()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "ca58a512-76a9-4e84-9106-e07739c29051",
   "metadata": {},
   "outputs": [],
   "source": [
    "transformThreadCount = 3\n",
    "logFileName = \"TableETL_KippFoundation.log\"\n",
    "sqlSchema = 'etl'\n",
    "saveFileBaseName = \"SalesForceEduCloud_KippFound_%s.csv\"\n",
    "outputTableBaseName = \"SalesForceEduCloud_KippFound_%s\"\n",
    "jsonConfigFileName = \"TableETL-KippFoundation.json\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "12604f21-59ac-4bee-9640-ac5e464edc68",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2024-08-26 14:22:05.935425] (MainThread) Started...\n"
     ]
    }
   ],
   "source": [
    "logQueue = Queue()\n",
    "\n",
    "\n",
    "def lprint(val):\n",
    "    outstr = f\"[{datetime.now()}] ({threading.current_thread().name}) {val}\"\n",
    "    print(outstr)\n",
    "    logQueue.put(outstr)\n",
    "    \n",
    "    \n",
    "lprint(\"Started...\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "10a1d393-b787-4854-b936-809071c1df42",
   "metadata": {},
   "outputs": [],
   "source": [
    "handleThreadError_Super = threading.excepthook\n",
    "\n",
    "def handleThreadError(args):\n",
    "    handleThreadError_Super(args)\n",
    "    lprint(f\"ERROR!!! {args.exc_type} \\\"{args.exc_value}\\\" in thread {args.thread}  ERROR!!!\")\n",
    "    \n",
    "threading.excepthook = handleThreadError"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "18ed47d5-80a5-4abf-8fce-c86235abe4e5",
   "metadata": {},
   "outputs": [],
   "source": [
    "def lprintDaemon():\n",
    "    with open(logFileName,\"w\") as logfile:\n",
    "        while True:\n",
    "            logfile.write(logQueue.get())\n",
    "            logfile.write(\"\\n\")\n",
    "            logfile.flush()\n",
    "            logQueue.task_done()\n",
    "\n",
    "lprintDaemonThread = threading.Thread(target=lprintDaemon)\n",
    "lprintDaemonThread.daemon = True\n",
    "lprintDaemonThread.name = \"lprintDaemon\"\n",
    "lprintDaemonThread.start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "e33cc6a3-2013-42c1-81bc-098f40c807f8",
   "metadata": {},
   "outputs": [],
   "source": [
    "#sfusername = environ.get('KFsfusername')\n",
    "#sfpassword = environ.get('KFsfpassword')\n",
    "#sfsecret = environ.get(\"KFsfsecret\")\n",
    "sfclientid = environ.get(\"KFsfclientid\")\n",
    "sfclientsecret = environ.get(\"KFsfclientsecret\")\n",
    "sfinstanceurl = environ.get(\"KFsfinstanceurl\")\n",
    "connstr = environ.get(\"KNOS_Datawarehouse\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "5dd7a590-da82-4936-ab95-0b4c94b4c2cf",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2024-08-26 14:22:05.974316] (MainThread) Creating engine\n"
     ]
    }
   ],
   "source": [
    "lprint(\"Creating engine\")\n",
    "engine = sa.create_engine(connstr, fast_executemany=True, isolation_level=\"READ UNCOMMITTED\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "d5313148-4902-43a4-a030-ab3a18e6cadf",
   "metadata": {},
   "outputs": [],
   "source": [
    "jsonDecoder = JSONDecoder()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "38316999-4360-4d79-ae30-c7a7ca230b4e",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2024-08-26 14:22:06.057445] (MainThread) Reading config file TableETL-KippFoundation.json\n",
      "[2024-08-26 14:22:06.059971] (MainThread) {'Tables': {'Account': {'Columns': ['Id', 'IsDeleted', 'MasterRecordId', 'Name', 'Type', 'RecordTypeId', 'ParentId', 'BillingPostalCode', 'Metro_Area__c', 'College_PowerRanking__c', 'DEP_College_PowerRanking__c', 'Long_Account_ID__c']}, 'Application__c': {'Columns': ['Id', 'IsDeleted', 'Name', 'RecordTypeId', 'Applicant__c', 'School__c', 'GPA_Type__c', 'GPA__c']}, 'Contact': {'Columns': ['Id', 'IsDeleted', 'AccountId', 'LastName', 'FirstName', 'Name', 'RecordTypeId', 'MailingPostalCode', 'Birthdate', 'Currently_Enrolled_School__c', 'High_School_Graduated_From__c', 'KIPP_Region_School__c', 'KCA_KIPP_ID__c', 'Salesforce_ID__c', 'College_Match_Display_GPA__c', 'Most_Recent_College_Application__c', 'Most_Recent_College_Enrollment_Status__c', 'Most_Recent_College_Enrollment__c']}, 'Enrollment__c': {'Columns': ['Id', 'IsDeleted', 'Name', 'RecordTypeId', 'Student__c', 'School__c', 'Student_HS_Cohort__c', 'Long_Enrollment_ID__c']}, 'Standardized_Test__c': {'Columns': ['Id', 'IsDeleted', 'Name', 'RecordTypeId', 'Contact__c', 'ACT_English__c', 'ACT_Math__c', 'ACT_Reading__c', 'ACT_Science__c']}}}\n"
     ]
    }
   ],
   "source": [
    "lprint(f\"Reading config file {jsonConfigFileName}\")\n",
    "with open(jsonConfigFileName) as f:\n",
    "\n",
    "    jsonConfig = jsonDecoder.decode( f.read() )\n",
    "\n",
    "lprint(jsonConfig)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "1e11a31c-e337-49d1-b7e2-2a4ef908bcac",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2024-08-26 14:22:06.072781] (MainThread) Setting destired tables\n",
      "[2024-08-26 14:22:06.075848] (MainThread) ['Account', 'Application__c', 'Contact', 'Enrollment__c', 'Standardized_Test__c']\n"
     ]
    }
   ],
   "source": [
    "lprint(\"Setting destired tables\")\n",
    "desiredTables = list(jsonConfig['Tables'].keys())\n",
    "lprint(desiredTables)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "ea24398c-41c7-44b2-94b0-e788406df531",
   "metadata": {},
   "outputs": [],
   "source": [
    "#soqlFilters = defaultdict(lambda: \"where IsDeleted = false LIMIT 1000\", {})\n",
    "soqlFilters = defaultdict(lambda: \"where IsDeleted = false\", {})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "15749657-dbd0-4041-af7f-4d45f97d6d48",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2024-08-26 14:22:06.108475] (MainThread) Getting access token for 3MVG9CVKiXR7Ri5rJHLGnofNMvaL5BIBQWSC4w5tfDVGFdeHsnNQ4swJ40eKiZ_mO8F0.9Rl_1MKToWdgbJ9J from instance https://kippadb.my.salesforce.com/\n",
      "[2024-08-26 14:22:06.109912] (MainThread) Getting Auth token from https://kippadb.my.salesforce.com/services/oauth2/token\n"
     ]
    }
   ],
   "source": [
    "lprint(f\"Getting access token for {sfclientid} from instance {sfinstanceurl}\")\n",
    "sfVersion='58.0'\n",
    "payload = {\n",
    "    'grant_type':'client_credentials',\n",
    "    'client_id':sfclientid,\n",
    "    'client_secret':sfclientsecret,\n",
    "    #'username':sfusername,\n",
    "    #'password':sfpassword+sfsecret\n",
    "}\n",
    "\n",
    "authURL = f\"{sfinstanceurl}services/oauth2/token\"\n",
    "#authURL = \"http://127.0.0.1:55555/services/oauth2/token\"\n",
    "lprint(f\"Getting Auth token from {authURL}\")\n",
    "session = requests.Session()\n",
    "authResp = session.post(authURL,\\\n",
    "                    data=payload,)\n",
    "\n",
    "authRespData = jsonDecoder.decode(authResp.text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "33788a09-daaf-465a-a9e9-4eac32fe0e38",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2024-08-26 14:22:08.042105] (MainThread) {'access_token': '****************************************************************************************************************', 'signature': '698OvwKXIiddY/ZIZfAuxnjbWIOqc5rAcsOeUD2Pglk=', 'scope': 'sfap_api api', 'instance_url': 'https://kippadb.my.salesforce.com', 'id': 'https://login.salesforce.com/id/00D80000000dTInEAM/005Qg000007xMJpIAM', 'token_type': 'Bearer', 'issued_at': '1724700128919', 'api_instance_url': 'https://api.salesforce.com'}\n"
     ]
    }
   ],
   "source": [
    "authRespDataPublic = authRespData.copy()\n",
    "\n",
    "if 'access_token' in authRespDataPublic.keys():\n",
    "    authRespDataPublic['access_token'] = '*' * len(authRespDataPublic['access_token'])\n",
    "\n",
    "lprint(authRespDataPublic)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "2ac22aa4-7509-4bf9-bfb2-98a4464074c3",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2024-08-26 14:22:08.064114] (MainThread) Creating Simple Salesforce Instance using sessionID...\n",
      "[2024-08-26 14:22:08.066114] (MainThread) Session created!\n"
     ]
    }
   ],
   "source": [
    "lprint(\"Creating Simple Salesforce Instance using sessionID...\")\n",
    "sf = Salesforce(instance_url=authRespData['instance_url'], session_id=authRespData['access_token'], version='58.0')\n",
    "lprint(\"Session created!\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "bb380df4-a9ab-4a38-ade5-18a5d861f0fe",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2024-08-26 14:22:08.081561] (MainThread) Getting metadata for Account\n",
      "[2024-08-26 14:22:10.352291] (MainThread) Getting metadata for Application__c\n",
      "[2024-08-26 14:22:10.666320] (MainThread) Getting metadata for Contact\n",
      "[2024-08-26 14:22:11.089200] (MainThread) Getting metadata for Enrollment__c\n",
      "[2024-08-26 14:22:11.357224] (MainThread) Getting metadata for Standardized_Test__c\n"
     ]
    }
   ],
   "source": [
    "metaData = {}\n",
    "metaDataLock = threading.Lock()\n",
    "\n",
    "for tbl in desiredTables:\n",
    "    fieldDescs = {}\n",
    "    \n",
    "    lprint(\"Getting metadata for %s\" % tbl)\n",
    "    \n",
    "    tblDesc = getattr(sf, tbl).describe()\n",
    "    \n",
    "    for field in tblDesc['fields']:\n",
    "        fieldDescs[field['name']] = {\n",
    "                                'type':field['type'],\n",
    "                                'length':field['length']\n",
    "        }\n",
    "        \n",
    "    \n",
    "    metaData[tbl] = fieldDescs\n",
    "    \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "922988fc-26b9-45c6-8895-ff3a26ddb98a",
   "metadata": {},
   "outputs": [],
   "source": [
    "daemonStatusLock = threading.Lock()\n",
    "daemonStatus = {}\n",
    "daemonCurTaskQueue = Queue()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "f574eff1-f9f8-4064-bcec-136f4415da11",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2024-08-26 14:22:11.608753] (MainThread) Creating extractDaemon Thread\n"
     ]
    }
   ],
   "source": [
    "outputDataQueue = Queue()\n",
    "extractDaemonStats = {}\n",
    "\n",
    "def extractDaemon():\n",
    "    daemonName = 'extractDaemon'\n",
    "    while True:\n",
    "        with daemonStatusLock:\n",
    "            daemonStatus[daemonName] = f\"{daemonName} waiting for job\"\n",
    "        \n",
    "        tbl = tableQueue.get()\n",
    "        daemonCurTaskQueue.put(daemonName)\n",
    "        \n",
    "        lprint(f\"Querying data for {tbl}\")\n",
    "        startTime = datetime.now()\n",
    "        \n",
    "        with daemonStatusLock:\n",
    "            daemonStatus[daemonName] = f\"{daemonName} Querying data for {tbl}\"\n",
    "\n",
    "        #with metaDataLock:\n",
    "        #    feilds = \", \".join(metaData[tbl].keys())\n",
    "\n",
    "        feilds = \", \".join(jsonConfig['Tables'][tbl]['Columns'])\n",
    "        \n",
    "        soql = \"select %s from %s %s\" % (feilds, tbl, soqlFilters[tbl])\n",
    "\n",
    "        \n",
    "        lprint(\"Starting query:  %s\" % soql)\n",
    "        resp = sf.query_all(soql)\n",
    "    \n",
    "        lprint(f\"Adding {tbl} to output queue\")\n",
    "        \n",
    "        lprint(\"Finished %s\" % tbl)\n",
    "    \n",
    "        lprint(\"%s totalRecords %d\" % (tbl, resp['totalSize']))\n",
    "        \n",
    "        outputDataQueue.put( (tbl, resp) )\n",
    "\n",
    "        extractDaemonStats[tbl] = datetime.now() - startTime\n",
    "        tableQueue.task_done()\n",
    "        daemonCurTaskQueue.get()\n",
    "        daemonCurTaskQueue.task_done()\n",
    "        \n",
    "lprint(\"Creating extractDaemon Thread\")\n",
    "extractDaemonThread = threading.Thread(target=extractDaemon)\n",
    "extractDaemonThread.daemon=True\n",
    "extractDaemonThread.name = 'extractDaemon'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "eba4f737-ea66-4509-bbe1-8d564d5ef77c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2024-08-26 14:22:11.637706] (MainThread) Creating dataFrameDaemon[0] thread\n",
      "[2024-08-26 14:22:11.638728] (MainThread) Creating dataFrameDaemon[1] thread\n",
      "[2024-08-26 14:22:11.638728] (MainThread) Creating dataFrameDaemon[2] thread\n"
     ]
    }
   ],
   "source": [
    "dataFramesQueue = Queue()\n",
    "transformDaemonStats = {}\n",
    "transformDaemonStatsLock = threading.Lock()\n",
    "transformDaemonThreads = []\n",
    "\n",
    "def transformDaemon(num):\n",
    "    daemonName = f\"transformDaemon[{num}]\"\n",
    "    while True:\n",
    "        with daemonStatusLock:\n",
    "            daemonStatus[daemonName] = f\"{daemonName} waiting for job\"\n",
    "            \n",
    "        (tbl, resp) = outputDataQueue.get()\n",
    "        daemonCurTaskQueue.put(daemonName)\n",
    "        \n",
    "        lprint(f\"Turning {tbl} into a dataframe\")\n",
    "        startTime = datetime.now()\n",
    "        \n",
    "        with daemonStatusLock:\n",
    "            daemonStatus[daemonName] = f\"{daemonName} Turning {tbl} into a dataframe\"\n",
    "        \n",
    "        respDf = pd.DataFrame.from_dict(resp['records'])\n",
    "    \n",
    "        if 'attributes' in respDf.columns:\n",
    "            \n",
    "            lprint(\"Dropping attributes from %s\" % tbl)\n",
    "            respDf.drop(columns=['attributes'], inplace=True)\n",
    "        \n",
    "        lprint(\"%s shape %s\" % (tbl, respDf.shape))\n",
    "\n",
    "\n",
    "        csvTblName = saveFileBaseName % tbl\n",
    "        lprint(f\"Saving {tbl} to {csvTblName}\")\n",
    "\n",
    "        with daemonStatusLock:\n",
    "            daemonStatus[daemonName] = f\"{daemonName} Saving {tbl} to {csvTblName}\"\n",
    "\n",
    "        respDf.to_csv(csvTblName, index=False)\n",
    "\n",
    "        lprint(\"Finished saving %s!\" % tbl)\n",
    "        \n",
    "        lprint(f\"transforming columns for {tbl}...\")\n",
    "\n",
    "        for col in respDf.columns:\n",
    "            with metaDataLock:\n",
    "                theType = metaData[tbl][col]['type']\n",
    "\n",
    "            with daemonStatusLock:\n",
    "                daemonStatus[daemonName] = f\"{daemonName} processing {tbl}->{col}...\"\n",
    "            \n",
    "            if theType in ['date', 'datetime']:\n",
    "                lprint(f\"Converting {tbl}->{col} to datetime...\")\n",
    "                respDf[col] = pd.to_datetime(respDf[col], errors='coerce')\n",
    "            \n",
    "    \n",
    "            for col in respDf.columns:\n",
    "        \n",
    "                if np.count_nonzero(respDf[col].map(lambda a: isinstance(a, dict ) or isinstance(a, list ))) > 0:\n",
    "                    lprint(\"Ordered dict found in %s column %s converting...\" % (tbl, col)) \n",
    "                    \n",
    "                    respDf[col] = respDf[col].map(lambda v: jcoder.encode(v) if not pd.isna(v) else None)\n",
    "                    \n",
    "                    lprint(\"Updating type for %s column %s to JSON...\" % (tbl, col)) \n",
    "                    with metaDataLock:\n",
    "                        metaData[tbl][col]['type'] = 'JSON'\n",
    "                    \n",
    "                    fieldLen = int(respDf[col].str.len().max())\n",
    "                    \n",
    "                    lprint(\"Setting fieldlen to %d for %s\" % (fieldLen, col))            \n",
    "                    with metaDataLock:\n",
    "                        metaData[tbl][col]['length'] = int(respDf[col].str.len().max())\n",
    "            \n",
    "\n",
    "        lprint(f\"Adding response for {tbl} to dataFrame queue\")\n",
    "\n",
    "        dataFramesQueue.put( (tbl, respDf) )\n",
    "        \n",
    "        with transformDaemonStatsLock:\n",
    "            transformDaemonStats[tbl] = datetime.now() - startTime\n",
    "            \n",
    "        outputDataQueue.task_done()\n",
    "        daemonCurTaskQueue.get()\n",
    "        daemonCurTaskQueue.task_done()\n",
    "\n",
    "for i in range(transformThreadCount):\n",
    "    lprint(f\"Creating dataFrameDaemon[{i}] thread\")\n",
    "    threadObj = threading.Thread(target=transformDaemon,args=(i,))\n",
    "    threadObj.daemon=True\n",
    "    threadObj.name = f\"transformDaemon[{i}]\"\n",
    "    transformDaemonThreads.append(threadObj)\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "7c015ba3-1be9-457e-9f5e-e24ed5ef5894",
   "metadata": {},
   "outputs": [],
   "source": [
    "jcoder = JSONEncoder()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "499adf2e-1cc8-4506-a642-9be75e28f1cf",
   "metadata": {},
   "outputs": [],
   "source": [
    "staticFields = {\n",
    "                 'boolean':sa.Boolean,\n",
    "                 'date':sa.DATE,\n",
    "                 'datetime':sa.DATETIME,\n",
    "                 'double': sa.FLOAT,\n",
    "                 #'email',\n",
    "                 #'id',\n",
    "                 'int':sa.INT,\n",
    "                 #'multipicklist',\n",
    "                 #'picklist',\n",
    "                 #'reference',\n",
    "                 #'string',\n",
    "                 'textarea':sa.TEXT,\n",
    "                 #'JSON':sa.JSON\n",
    "}\n",
    "\n",
    "#this is a mess\n",
    "def getSQLTypes(tbl, respDf):\n",
    "    with metaDataLock:\n",
    "        sqlTypes = {}\n",
    "        \n",
    "        lprint(\"Getting SQLTypes for %s\" % tbl)\n",
    "        \n",
    "        curMeta = metaData[tbl]\n",
    "            \n",
    "        for field in jsonConfig['Tables'][tbl]['Columns']:\n",
    "\n",
    "            \n",
    "            \n",
    "            if curMeta[field]['type'] in staticFields.keys():\n",
    "                sqlTypes[field] = staticFields[curMeta[field]['type']]()\n",
    "            \n",
    "            \n",
    "            else:\n",
    "                fieldLen = curMeta[field]['length']\n",
    "                \n",
    "                if fieldLen <= 255:\n",
    "                    sqlTypes[field] = sa.NVARCHAR(fieldLen)\n",
    "                \n",
    "                #this is a fix they set some of the custom field max values to weird stuff\n",
    "                elif np.count_nonzero(~pd.isna(respDf[field])) > 0 \\\n",
    "                                and ( fieldLen := int(respDf[field].str.len().max())) <= 255:\n",
    "                    sqlTypes[field] = sa.NVARCHAR(fieldLen)                \n",
    "                    \n",
    "                else:\n",
    "                    sqlTypes[field] = sa.TEXT()\n",
    "                \n",
    "    return sqlTypes\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "13833977-2514-4465-8e4b-e2cadbf061c8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2024-08-26 14:22:11.746218] (MainThread) Creating loadDaemon thread\n"
     ]
    }
   ],
   "source": [
    "loadDaemonStats = {}\n",
    "def loadDaemon():\n",
    "    daemonName = 'loadDaemon'\n",
    "    while True:\n",
    "        with daemonStatusLock:\n",
    "            daemonStatus[daemonName] = f\"{daemonName} waiting for job\"\n",
    "        \n",
    "        (tbl, respDf) = dataFramesQueue.get()\n",
    "        daemonCurTaskQueue.put(daemonName)\n",
    "        \n",
    "        startTime = datetime.now()\n",
    "        \n",
    "        if respDf.shape[0] == 0:\n",
    "            lprint(\"Skipping %s no data!\" % tbl)\n",
    "            loadDaemonStats[tbl] = \"Skipped, no data!\"\n",
    "            dataFramesQueue.task_done()\n",
    "            daemonCurTaskQueue.get()\n",
    "            daemonCurTaskQueue.task_done()\n",
    "            continue\n",
    "\n",
    "        with daemonStatusLock:\n",
    "            daemonStatus[daemonName] = f\"{daemonName} getting SQL types for {tbl}\"\n",
    "            \n",
    "        sqlTypes = getSQLTypes(tbl, respDf)\n",
    "        \n",
    "        \n",
    "        sqlTblName = outputTableBaseName % tbl\n",
    "        lprint(\"Uploading table %s to etl.%s\" % (tbl,  sqlTblName))\n",
    "\n",
    "        with daemonStatusLock:\n",
    "            daemonStatus[daemonName] = f\"{daemonName} uploading {tbl}\"\n",
    "            \n",
    "        with engine.begin() as conn:\n",
    "            respDf.to_sql(sqlTblName, conn, schema=sqlSchema, if_exists='replace', index=False, dtype=sqlTypes, chunksize=1)\n",
    "        \n",
    "        lprint(\"Finished uploading %s!\" % tbl)\n",
    "        loadDaemonStats[tbl] = datetime.now() - startTime\n",
    "\n",
    "        dataFramesQueue.task_done()\n",
    "        daemonCurTaskQueue.get()\n",
    "        daemonCurTaskQueue.task_done()\n",
    "\n",
    "lprint(\"Creating loadDaemon thread\")\n",
    "loadDaemonThread = threading.Thread(target=loadDaemon)\n",
    "loadDaemonThread.daemon=True\n",
    "loadDaemonThread.name = \"loadDaemon\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "d2444655-b79e-4a06-b44c-cef756c3c01e",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2024-08-26 14:22:11.783764] (MainThread) Queuing Account\n",
      "[2024-08-26 14:22:11.783764] (MainThread) Queuing Application__c\n",
      "[2024-08-26 14:22:11.783764] (MainThread) Queuing Contact\n",
      "[2024-08-26 14:22:11.783764] (MainThread) Queuing Enrollment__c\n",
      "[2024-08-26 14:22:11.783764] (MainThread) Queuing Standardized_Test__c\n"
     ]
    }
   ],
   "source": [
    "tableQueue = Queue()\n",
    "for tbl in desiredTables:\n",
    "    lprint(f\"Queuing {tbl}\")\n",
    "    tableQueue.put(tbl)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "0b986c73-b554-4d4e-96f6-11f76312aecc",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2024-08-26 14:22:11.810191] (MainThread) Starting loadDaemon thread\n",
      "[2024-08-26 14:22:12.477740] (loadDaemon) Getting SQLTypes for Account\n",
      "[2024-08-26 14:22:12.483989] (loadDaemon) Uploading table Account to etl.SalesForceEduCloud_KippFound_Account\n",
      "[2024-08-26 14:28:27.016036] (loadDaemon) Finished uploading Account!\n",
      "[2024-08-26 14:28:27.019037] (loadDaemon) Getting SQLTypes for Application__c\n",
      "[2024-08-26 14:28:27.019037] (loadDaemon) Uploading table Application__c to etl.SalesForceEduCloud_KippFound_Application__c\n",
      "[2024-08-26 14:30:29.089296] (loadDaemon) Finished uploading Application__c!\n",
      "[2024-08-26 14:30:29.090303] (loadDaemon) Getting SQLTypes for Contact\n",
      "[2024-08-26 14:30:29.094973] (loadDaemon) Uploading table Contact to etl.SalesForceEduCloud_KippFound_Contact\n",
      "[2024-08-26 14:35:47.988536] (loadDaemon) Finished uploading Contact!\n",
      "[2024-08-26 14:35:47.988536] (loadDaemon) Getting SQLTypes for Enrollment__c\n",
      "[2024-08-26 14:35:48.000101] (loadDaemon) Uploading table Enrollment__c to etl.SalesForceEduCloud_KippFound_Enrollment__c\n",
      "[2024-08-26 14:38:07.502623] (loadDaemon) Finished uploading Enrollment__c!\n",
      "[2024-08-26 14:38:07.503630] (loadDaemon) Getting SQLTypes for Standardized_Test__c\n",
      "[2024-08-26 14:38:07.503630] (loadDaemon) Uploading table Standardized_Test__c to etl.SalesForceEduCloud_KippFound_Standardized_Test__c\n",
      "[2024-08-26 14:41:31.229546] (loadDaemon) Finished uploading Standardized_Test__c!\n"
     ]
    }
   ],
   "source": [
    "lprint(\"Starting loadDaemon thread\")\n",
    "loadDaemonThread.start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "fe12092c-1883-4e1e-b7f6-f24a6779da0b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2024-08-26 14:22:11.832577] (MainThread) Starting dataFrameDaemon threads\n",
      "[2024-08-26 14:22:11.832577] (MainThread) Starting dataFrameDaemon[0]\n",
      "[2024-08-26 14:22:11.834845] (MainThread) Starting dataFrameDaemon[1]\n",
      "[2024-08-26 14:22:11.835920] (MainThread) Starting dataFrameDaemon[2]\n",
      "[2024-08-26 14:22:12.345455] (transformDaemon[0]) Turning Account into a dataframe\n",
      "[2024-08-26 14:22:12.356323] (transformDaemon[0]) Dropping attributes from Account\n",
      "[2024-08-26 14:22:12.357285] (transformDaemon[0]) Account shape (1000, 12)\n",
      "[2024-08-26 14:22:12.357285] (transformDaemon[0]) Saving Account to SalesForceEduCloud_KippFound_Account.csv\n",
      "[2024-08-26 14:22:12.378266] (transformDaemon[0]) Finished saving Account!\n",
      "[2024-08-26 14:22:12.378266] (transformDaemon[0]) transforming columns for Account...\n",
      "[2024-08-26 14:22:12.477740] (transformDaemon[0]) Adding response for Account to dataFrame queue\n",
      "[2024-08-26 14:22:13.217810] (transformDaemon[1]) Turning Application__c into a dataframe\n",
      "[2024-08-26 14:22:13.228856] (transformDaemon[1]) Dropping attributes from Application__c\n",
      "[2024-08-26 14:22:13.230836] (transformDaemon[1]) Application__c shape (1000, 8)\n",
      "[2024-08-26 14:22:13.231836] (transformDaemon[1]) Saving Application__c to SalesForceEduCloud_KippFound_Application__c.csv\n",
      "[2024-08-26 14:22:13.250384] (transformDaemon[1]) Finished saving Application__c!\n",
      "[2024-08-26 14:22:13.250384] (transformDaemon[1]) transforming columns for Application__c...\n",
      "[2024-08-26 14:22:13.326697] (transformDaemon[1]) Adding response for Application__c to dataFrame queue\n",
      "[2024-08-26 14:22:13.700288] (transformDaemon[2]) Turning Contact into a dataframe\n",
      "[2024-08-26 14:22:13.772445] (transformDaemon[2]) Dropping attributes from Contact\n",
      "[2024-08-26 14:22:13.773454] (transformDaemon[2]) Contact shape (1000, 18)\n",
      "[2024-08-26 14:22:13.773454] (transformDaemon[2]) Saving Contact to SalesForceEduCloud_KippFound_Contact.csv\n",
      "[2024-08-26 14:22:13.790994] (transformDaemon[2]) Finished saving Contact!\n",
      "[2024-08-26 14:22:13.790994] (transformDaemon[2]) transforming columns for Contact...\n",
      "[2024-08-26 14:22:13.889752] (transformDaemon[2]) Converting Contact->Birthdate to datetime...\n",
      "[2024-08-26 14:22:14.033156] (transformDaemon[2]) Adding response for Contact to dataFrame queue\n",
      "[2024-08-26 14:22:14.211110] (transformDaemon[0]) Turning Enrollment__c into a dataframe\n",
      "[2024-08-26 14:22:14.218069] (transformDaemon[0]) Dropping attributes from Enrollment__c\n",
      "[2024-08-26 14:22:14.218069] (transformDaemon[0]) Enrollment__c shape (1000, 8)\n",
      "[2024-08-26 14:22:14.218069] (transformDaemon[0]) Saving Enrollment__c to SalesForceEduCloud_KippFound_Enrollment__c.csv\n",
      "[2024-08-26 14:22:14.232223] (transformDaemon[0]) Finished saving Enrollment__c!\n",
      "[2024-08-26 14:22:14.232223] (transformDaemon[0]) transforming columns for Enrollment__c...\n",
      "[2024-08-26 14:22:14.269931] (transformDaemon[0]) Adding response for Enrollment__c to dataFrame queue\n",
      "[2024-08-26 14:22:14.582348] (transformDaemon[1]) Turning Standardized_Test__c into a dataframe\n",
      "[2024-08-26 14:22:14.590385] (transformDaemon[1]) Dropping attributes from Standardized_Test__c\n",
      "[2024-08-26 14:22:14.590385] (transformDaemon[1]) Standardized_Test__c shape (1000, 9)\n",
      "[2024-08-26 14:22:14.590385] (transformDaemon[1]) Saving Standardized_Test__c to SalesForceEduCloud_KippFound_Standardized_Test__c.csv\n",
      "[2024-08-26 14:22:14.599486] (transformDaemon[1]) Finished saving Standardized_Test__c!\n",
      "[2024-08-26 14:22:14.599486] (transformDaemon[1]) transforming columns for Standardized_Test__c...\n",
      "[2024-08-26 14:22:14.646091] (transformDaemon[1]) Adding response for Standardized_Test__c to dataFrame queue\n"
     ]
    }
   ],
   "source": [
    "lprint(\"Starting dataFrameDaemon threads\")\n",
    "for i,thread in enumerate(transformDaemonThreads):\n",
    "    lprint(f\"Starting dataFrameDaemon[{i}]\")\n",
    "    thread.start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "34ce58d4-156f-49e5-a14f-2f470ae08dde",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2024-08-26 14:22:11.848577] (MainThread) Starting query Thread\n",
      "[2024-08-26 14:22:11.849577] (extractDaemon) Querying data for Account\n",
      "[2024-08-26 14:22:11.849577] (extractDaemon) Starting query:  select Id, IsDeleted, MasterRecordId, Name, Type, RecordTypeId, ParentId, BillingPostalCode, Metro_Area__c, College_PowerRanking__c, DEP_College_PowerRanking__c, Long_Account_ID__c from Account where IsDeleted = false LIMIT 1000\n",
      "[2024-08-26 14:22:12.342220] (extractDaemon) Adding Account to output queue\n",
      "[2024-08-26 14:22:12.345455] (extractDaemon) Finished Account\n",
      "[2024-08-26 14:22:12.345455] (extractDaemon) Account totalRecords 1000\n",
      "[2024-08-26 14:22:12.345455] (extractDaemon) Querying data for Application__c\n",
      "[2024-08-26 14:22:12.345455] (extractDaemon) Starting query:  select Id, IsDeleted, Name, RecordTypeId, Applicant__c, School__c, GPA_Type__c, GPA__c from Application__c where IsDeleted = false LIMIT 1000\n",
      "[2024-08-26 14:22:13.215780] (extractDaemon) Adding Application__c to output queue\n",
      "[2024-08-26 14:22:13.216793] (extractDaemon) Finished Application__c\n",
      "[2024-08-26 14:22:13.216793] (extractDaemon) Application__c totalRecords 1000\n",
      "[2024-08-26 14:22:13.216793] (extractDaemon) Querying data for Contact\n",
      "[2024-08-26 14:22:13.216793] (extractDaemon) Starting query:  select Id, IsDeleted, AccountId, LastName, FirstName, Name, RecordTypeId, MailingPostalCode, Birthdate, Currently_Enrolled_School__c, High_School_Graduated_From__c, KIPP_Region_School__c, KCA_KIPP_ID__c, Salesforce_ID__c, College_Match_Display_GPA__c, Most_Recent_College_Application__c, Most_Recent_College_Enrollment_Status__c, Most_Recent_College_Enrollment__c from Contact where IsDeleted = false LIMIT 1000\n",
      "[2024-08-26 14:22:13.700288] (extractDaemon) Adding Contact to output queue\n",
      "[2024-08-26 14:22:13.700288] (extractDaemon) Finished Contact\n",
      "[2024-08-26 14:22:13.700288] (extractDaemon) Contact totalRecords 1000\n",
      "[2024-08-26 14:22:13.700288] (extractDaemon) Querying data for Enrollment__c\n",
      "[2024-08-26 14:22:13.700288] (extractDaemon) Starting query:  select Id, IsDeleted, Name, RecordTypeId, Student__c, School__c, Student_HS_Cohort__c, Long_Enrollment_ID__c from Enrollment__c where IsDeleted = false LIMIT 1000\n",
      "[2024-08-26 14:22:14.210062] (extractDaemon) Adding Enrollment__c to output queue\n",
      "[2024-08-26 14:22:14.211110] (extractDaemon) Finished Enrollment__c\n",
      "[2024-08-26 14:22:14.211110] (extractDaemon) Enrollment__c totalRecords 1000\n",
      "[2024-08-26 14:22:14.211110] (extractDaemon) Querying data for Standardized_Test__c\n",
      "[2024-08-26 14:22:14.211110] (extractDaemon) Starting query:  select Id, IsDeleted, Name, RecordTypeId, Contact__c, ACT_English__c, ACT_Math__c, ACT_Reading__c, ACT_Science__c from Standardized_Test__c where IsDeleted = false LIMIT 1000\n",
      "[2024-08-26 14:22:14.573664] (extractDaemon) Adding Standardized_Test__c to output queue\n",
      "[2024-08-26 14:22:14.573664] (extractDaemon) Finished Standardized_Test__c\n",
      "[2024-08-26 14:22:14.573664] (extractDaemon) Standardized_Test__c totalRecords 1000\n"
     ]
    }
   ],
   "source": [
    "lprint(\"Starting query Thread\")\n",
    "extractDaemonThread.start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "3900cebd-2985-46d5-baa1-3930d87af6c8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2024-08-26 14:22:11.870059] (MainThread) All daemons started!\n"
     ]
    }
   ],
   "source": [
    "lprint(\"All daemons started!\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "f60da6ec-61c0-427f-b00e-be17ee270155",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2024-08-26 14:22:11.891257] (MainThread) Daemon Status:\n",
      "loadDaemon waiting for job\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon Querying data for Account\n",
      "[2024-08-26 14:22:11.891257] (MainThread) Extract queue(4)\tTransform queue(0)\tLoad queue(0)\tRunning(1)\tTotal(5)\n",
      "[2024-08-26 14:22:41.892063] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Account\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:22:41.892063] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(4)\tRunning(1)\tTotal(5)\n",
      "[2024-08-26 14:23:11.892869] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Account\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:23:11.892869] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(4)\tRunning(1)\tTotal(5)\n",
      "[2024-08-26 14:23:41.893524] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Account\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:23:41.894053] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(4)\tRunning(1)\tTotal(5)\n",
      "[2024-08-26 14:24:11.894293] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Account\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:24:11.894293] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(4)\tRunning(1)\tTotal(5)\n",
      "[2024-08-26 14:24:41.895250] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Account\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:24:41.895250] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(4)\tRunning(1)\tTotal(5)\n",
      "[2024-08-26 14:25:11.895910] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Account\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:25:11.895910] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(4)\tRunning(1)\tTotal(5)\n",
      "[2024-08-26 14:25:41.896939] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Account\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:25:41.896939] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(4)\tRunning(1)\tTotal(5)\n",
      "[2024-08-26 14:26:11.897919] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Account\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:26:11.897919] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(4)\tRunning(1)\tTotal(5)\n",
      "[2024-08-26 14:26:41.898402] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Account\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:26:41.898402] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(4)\tRunning(1)\tTotal(5)\n",
      "[2024-08-26 14:27:11.899181] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Account\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:27:11.899181] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(4)\tRunning(1)\tTotal(5)\n",
      "[2024-08-26 14:27:41.900039] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Account\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:27:41.900039] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(4)\tRunning(1)\tTotal(5)\n",
      "[2024-08-26 14:28:11.900753] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Account\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:28:11.900753] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(4)\tRunning(1)\tTotal(5)\n",
      "[2024-08-26 14:28:41.901407] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Application__c\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:28:41.901407] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(3)\tRunning(1)\tTotal(4)\n",
      "[2024-08-26 14:29:11.902165] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Application__c\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:29:11.902165] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(3)\tRunning(1)\tTotal(4)\n",
      "[2024-08-26 14:29:41.902785] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Application__c\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:29:41.902785] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(3)\tRunning(1)\tTotal(4)\n",
      "[2024-08-26 14:30:11.903502] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Application__c\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:30:11.903502] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(3)\tRunning(1)\tTotal(4)\n",
      "[2024-08-26 14:30:41.904261] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Contact\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:30:41.904261] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(2)\tRunning(1)\tTotal(3)\n",
      "[2024-08-26 14:31:11.904688] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Contact\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:31:11.904688] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(2)\tRunning(1)\tTotal(3)\n",
      "[2024-08-26 14:31:41.905127] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Contact\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:31:41.906674] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(2)\tRunning(1)\tTotal(3)\n",
      "[2024-08-26 14:32:11.907245] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Contact\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:32:11.907245] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(2)\tRunning(1)\tTotal(3)\n",
      "[2024-08-26 14:32:41.907892] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Contact\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:32:41.907892] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(2)\tRunning(1)\tTotal(3)\n",
      "[2024-08-26 14:33:11.908401] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Contact\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:33:11.908401] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(2)\tRunning(1)\tTotal(3)\n",
      "[2024-08-26 14:33:41.909188] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Contact\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:33:41.909188] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(2)\tRunning(1)\tTotal(3)\n",
      "[2024-08-26 14:34:11.909607] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Contact\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:34:11.909607] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(2)\tRunning(1)\tTotal(3)\n",
      "[2024-08-26 14:34:41.910245] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Contact\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:34:41.910245] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(2)\tRunning(1)\tTotal(3)\n",
      "[2024-08-26 14:35:11.910741] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Contact\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:35:11.910741] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(2)\tRunning(1)\tTotal(3)\n",
      "[2024-08-26 14:35:41.911238] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Contact\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:35:41.911238] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(2)\tRunning(1)\tTotal(3)\n",
      "[2024-08-26 14:36:11.911881] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Enrollment__c\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:36:11.911881] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(1)\tRunning(1)\tTotal(2)\n",
      "[2024-08-26 14:36:41.912823] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Enrollment__c\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:36:41.912823] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(1)\tRunning(1)\tTotal(2)\n",
      "[2024-08-26 14:37:11.913359] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Enrollment__c\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:37:11.913359] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(1)\tRunning(1)\tTotal(2)\n",
      "[2024-08-26 14:37:41.914432] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Enrollment__c\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:37:41.914432] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(1)\tRunning(1)\tTotal(2)\n",
      "[2024-08-26 14:38:11.915088] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Standardized_Test__c\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:38:11.915088] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(0)\tRunning(1)\tTotal(1)\n",
      "[2024-08-26 14:38:41.915831] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Standardized_Test__c\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:38:41.915831] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(0)\tRunning(1)\tTotal(1)\n",
      "[2024-08-26 14:39:11.916569] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Standardized_Test__c\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:39:11.916569] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(0)\tRunning(1)\tTotal(1)\n",
      "[2024-08-26 14:39:41.917466] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Standardized_Test__c\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:39:41.917466] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(0)\tRunning(1)\tTotal(1)\n",
      "[2024-08-26 14:40:11.918276] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Standardized_Test__c\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:40:11.918276] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(0)\tRunning(1)\tTotal(1)\n",
      "[2024-08-26 14:40:41.937278] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Standardized_Test__c\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:40:41.938300] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(0)\tRunning(1)\tTotal(1)\n",
      "[2024-08-26 14:41:11.939636] (MainThread) Daemon Status:\n",
      "loadDaemon uploading Standardized_Test__c\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:41:11.939636] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(0)\tRunning(1)\tTotal(1)\n",
      "[2024-08-26 14:41:41.940639] (MainThread) Daemon Status:\n",
      "loadDaemon waiting for job\n",
      "transformDaemon[0] waiting for job\n",
      "transformDaemon[1] waiting for job\n",
      "transformDaemon[2] waiting for job\n",
      "extractDaemon waiting for job\n",
      "[2024-08-26 14:41:41.940639] (MainThread) Extract queue(0)\tTransform queue(0)\tLoad queue(0)\tRunning(0)\tTotal(0)\n",
      "[2024-08-26 14:42:11.946156] (MainThread) All queues are empty but uploading is most likely still continueing\n"
     ]
    }
   ],
   "source": [
    "comboSize = 1\n",
    "while comboSize > 0:\n",
    "    tqSize = tableQueue.qsize()\n",
    "    odqSize = outputDataQueue.qsize()\n",
    "    dfqSize = dataFramesQueue.qsize()\n",
    "    runSize = daemonCurTaskQueue.qsize()\n",
    "    comboSize = tqSize + odqSize + dfqSize + runSize\n",
    "    \n",
    "    with daemonStatusLock:\n",
    "        lprint(\"Daemon Status:\\n\" + (\"\\n\".join(daemonStatus.values())))\n",
    "\n",
    "    lprint(f\"Extract queue({tqSize})\\tTransform queue({odqSize})\\tLoad queue({dfqSize})\\tRunning({runSize})\\tTotal({comboSize})\")\n",
    "        \n",
    "    sleep(30)\n",
    "\n",
    "lprint(\"All queues are empty but uploading is most likely still continueing\")\n",
    "    \n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "82f7bddb-5e21-430e-a6e6-77368c70447a",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2024-08-26 14:42:11.960131] (MainThread) All table data extracted\n",
      "[2024-08-26 14:42:11.960131] (MainThread) All data transformed!\n",
      "[2024-08-26 14:42:11.960131] (MainThread) All data loaded to Engine(mssql+pyodbc://gbezet:***@KNOS_Datawarehouse)\n"
     ]
    }
   ],
   "source": [
    "tableQueue.join()\n",
    "lprint(\"All table data extracted\")\n",
    "outputDataQueue.join()\n",
    "lprint(\"All data transformed!\")\n",
    "dataFramesQueue.join()\n",
    "lprint(f\"All data loaded to {engine}\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "00c1b961-ec39-42a8-ae29-acabcb030281",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2024-08-26 14:42:11.977758] (MainThread) All running daemons have completed!\n"
     ]
    }
   ],
   "source": [
    "daemonCurTaskQueue.join()\n",
    "lprint(\"All running daemons have completed!\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "6d602dba-4f95-49ed-97e9-d63f819396de",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2024-08-26 14:42:11.986280] (MainThread) extractDaemon processed Account in 0:00:00.495878\n",
      "[2024-08-26 14:42:11.986280] (MainThread) extractDaemon processed Application__c in 0:00:00.871338\n",
      "[2024-08-26 14:42:11.986280] (MainThread) extractDaemon processed Contact in 0:00:00.483495\n",
      "[2024-08-26 14:42:11.986280] (MainThread) extractDaemon processed Enrollment__c in 0:00:00.510822\n",
      "[2024-08-26 14:42:11.986280] (MainThread) extractDaemon processed Standardized_Test__c in 0:00:00.362554\n"
     ]
    }
   ],
   "source": [
    "for key in sorted(extractDaemonStats.keys()):\n",
    "    lprint(f\"extractDaemon processed {key} in {extractDaemonStats[key]}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "id": "2bb9f1d9-82c9-4935-99fa-6fad4b680431",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2024-08-26 14:42:12.006172] (MainThread) transformDaemon processed Account in 0:00:00.132285\n",
      "[2024-08-26 14:42:12.006172] (MainThread) transformDaemon processed Application__c in 0:00:00.108887\n",
      "[2024-08-26 14:42:12.006172] (MainThread) transformDaemon processed Contact in 0:00:00.334932\n",
      "[2024-08-26 14:42:12.006172] (MainThread) transformDaemon processed Enrollment__c in 0:00:00.062157\n",
      "[2024-08-26 14:42:12.006172] (MainThread) transformDaemon processed Standardized_Test__c in 0:00:00.063743\n"
     ]
    }
   ],
   "source": [
    "for key in sorted(transformDaemonStats.keys()):\n",
    "    lprint(f\"transformDaemon processed {key} in {transformDaemonStats[key]}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "a2fa619a-168f-4f84-9f36-c23e33c044a8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2024-08-26 14:42:12.015129] (MainThread) loadDaemon processed Account in 0:06:14.541297\n",
      "[2024-08-26 14:42:12.016133] (MainThread) loadDaemon processed Application__c in 0:02:02.070259\n",
      "[2024-08-26 14:42:12.016133] (MainThread) loadDaemon processed Contact in 0:05:18.898233\n",
      "[2024-08-26 14:42:12.016133] (MainThread) loadDaemon processed Enrollment__c in 0:02:19.514087\n",
      "[2024-08-26 14:42:12.016133] (MainThread) loadDaemon processed Standardized_Test__c in 0:03:23.726923\n"
     ]
    }
   ],
   "source": [
    "for key in sorted(loadDaemonStats.keys()):\n",
    "    lprint(f\"loadDaemon processed {key} in {loadDaemonStats[key]}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "6ced5748-e6e3-417c-96ab-23b04cf87b94",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2024-08-26 14:42:12.025432] (MainThread) Account completed in 0:06:15.169460\n",
      "[2024-08-26 14:42:12.025940] (MainThread) Application__c completed in 0:02:03.050484\n",
      "[2024-08-26 14:42:12.025940] (MainThread) Contact completed in 0:05:19.716660\n",
      "[2024-08-26 14:42:12.025940] (MainThread) Enrollment__c completed in 0:02:20.087066\n",
      "[2024-08-26 14:42:12.025940] (MainThread) Standardized_Test__c completed in 0:03:24.153220\n"
     ]
    }
   ],
   "source": [
    "statLists = [extractDaemonStats, transformDaemonStats, loadDaemonStats]\n",
    "for key in sorted(extractDaemonStats.keys()):\n",
    "    if np.all([isinstance(statList[key], timedelta) for statList in statLists]):\n",
    "        comboTime = np.sum([statList[key] for statList in statLists])\n",
    "        lprint(f\"{key} completed in {comboTime}\")\n",
    "    else:\n",
    "        lprint(F\"{key} incomplete {[str(l[key]) for l in statLists]}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "id": "a0674795-ed31-455b-ad19-6fdc45ec84b2",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2024-08-26 14:42:12.034590] (MainThread) =============DONE!===================\n"
     ]
    }
   ],
   "source": [
    "lprint(\"=============DONE!===================\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "id": "b7285457-e60b-40b6-8507-4713018baa69",
   "metadata": {},
   "outputs": [],
   "source": [
    "sleep(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "id": "bd05b391-0146-4bb5-826f-5e4c7c98046f",
   "metadata": {},
   "outputs": [],
   "source": [
    "#jupyter nbconvert .\\TableETL-KippFoundation.ipynb --to python"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0f985bc2-37f7-4c85-a301-4fbf977f01da",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}