04_DHS_Load_New_Tables.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Load data tables to the database\n",
    "\n",
    "This stage should be run only after the metadata have been updated in the database in stage 03.\n",
    "\n",
    "For each incoming data file, we identify which database table it is bound for. Based on the metadata tables, we ensure that this table exists, and that all the columns it should have exist, and that they are all wide enough. This only needs doing once per destination table, not once per file, but the helper code handles this so you can just loop over all the input files.  \n",
    "\n",
    "Note that this will create the tables, but not the schema: you must specify a data schema which already exists, and then tables will be created as needed.\n",
    "\n",
    "Now that the database is ready, for each datafile we check whether any data for this survey/table are already in the database. If so, we check if the number of rows matches. If so, then nothing is done. If not then we drop any existing data and (re)load from the file contents. Again this is done with a simple loop over all the parsed data CSV files.\n",
    "\n",
    "Some tables are stored with most of the data packed into a single JSONB column. These tables are those with more than a certain number of columns in total (currently 500), as well as those which are for country-specific data (these are likely to be very sparsely populated, with a growing number of columns). If the destination table is one of these, then the data columns will be packed into this single JSON attribute, whilst columns used for joining or indexing the table will be left as first-class.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "with open('config/pg_conn.txt') as conn_details:\n",
    "    conn_str_psyco = conn_details.readline()\n",
    "    conn_str_sqlalchemy = conn_details.readline()\n",
    "\n",
    "import pandas as pd\n",
    "import psycopg2 as pg\n",
    "from sqlalchemy import create_engine\n",
    "import os\n",
    "import glob"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from DHS_To_Database.lib04_Update_Table_Data import TableDataHelper"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "engine = create_engine(conn_str_sqlalchemy)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "SPEC_SCHEMA = 'dhs_survey_specs'\n",
    "DATA_SCHEMA = 'dhs_data_tables'\n",
    "\n",
    "TABLESPEC_TABLENAME = 'dhs_table_specs_flat'\n",
    "VALUESPEC_TABLENAME = 'dhs_value_descs'\n",
    "SURVEYLIST_TABLENAME = 'dhs_survey_listing'\n",
    "\n",
    "TABLE_SPEC_TABLE = \".\".join((SPEC_SCHEMA, TABLESPEC_TABLENAME))\n",
    "VALUE_SPEC_TABLE = \".\".join((SPEC_SCHEMA, VALUESPEC_TABLENAME))\n",
    "SURVEYLIST_TABLE = \".\".join((SPEC_SCHEMA, SURVEYLIST_TABLENAME))\n",
    "\n",
    "STAGING_FOLDER = \"/mnt/c/Users/harry/OneDrive - Nexus365/Informal_Cities/DHS_Data_And_Prep/Staging\"\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "_data_folder = os.path.join(STAGING_FOLDER, 'tables')\n",
    "data_files = glob.glob(os.path.join(_data_folder, \"*.csv\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data_files[:5]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "10804"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "len(data_files)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Initialise a TableDataHelper which will handle all the schema checks and data loading"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "db_helper = TableDataHelper(conn_str=conn_str_sqlalchemy, table_spec_table=TABLESPEC_TABLENAME,\n",
    "                             value_spec_table=VALUESPEC_TABLENAME, spec_schema=SPEC_SCHEMA,\n",
    "                           data_schema=DATA_SCHEMA, dry_run=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "db_helper._is_dry_run"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Uncomment and execute this when you're ready to ~~break things~~ run the updates"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "#db_helper._is_dry_run = False"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "db_helper.drop_and_reload('/mnt/c/Users/harry/OneDrive - Nexus365/Informal_Cities/DHS_Data_And_Prep/Staging/tables/156.IAIR42.RECH4.csv')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Prepare the database \n",
    "\n",
    "Check that all necessary tables exist and have the required columns. \n",
    "\n",
    "This is only done once for each distinct destination table (table_name), prepare_db_for_file is a no-op if it's already been done. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for table_file in data_files:\n",
    "    surveyid, _, _, _, table_name = TableDataHelper.parse_table_name(table_file)\n",
    "    # creates the table if it doesn't exist; otherwise \n",
    "    # checks that all required columns exist and are wide enough\n",
    "    # (compared to the metadata)   \n",
    "    db_helper.prepare_db_for_file(table_name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 74,
   "metadata": {
    "collapsed": true,
    "jupyter": {
     "outputs_hidden": true
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'MREC01',\n",
       " 'MREC11',\n",
       " 'MREC22',\n",
       " 'MREC31',\n",
       " 'MREC32',\n",
       " 'MREC41',\n",
       " 'MREC51',\n",
       " 'MREC61',\n",
       " 'MREC71',\n",
       " 'MREC75',\n",
       " 'MREC80',\n",
       " 'MREC83',\n",
       " 'MREC84',\n",
       " 'MREC85',\n",
       " 'MREC91',\n",
       " 'MREC92',\n",
       " 'MREC93',\n",
       " 'MREC94',\n",
       " 'MREC97',\n",
       " 'MREC98',\n",
       " 'MREC99',\n",
       " 'MRECDV',\n",
       " 'MRECFG',\n",
       " 'MRECGC',\n",
       " 'REC01',\n",
       " 'REC11',\n",
       " 'REC21',\n",
       " 'REC22',\n",
       " 'REC31',\n",
       " 'REC32',\n",
       " 'REC33',\n",
       " 'REC41',\n",
       " 'REC42',\n",
       " 'REC43',\n",
       " 'REC44',\n",
       " 'REC4A',\n",
       " 'REC51',\n",
       " 'REC61',\n",
       " 'REC71',\n",
       " 'REC75',\n",
       " 'REC80',\n",
       " 'REC81',\n",
       " 'REC82',\n",
       " 'REC83',\n",
       " 'REC84',\n",
       " 'REC85',\n",
       " 'REC91',\n",
       " 'REC92',\n",
       " 'REC93',\n",
       " 'REC94',\n",
       " 'REC95',\n",
       " 'REC96',\n",
       " 'REC97',\n",
       " 'REC98',\n",
       " 'REC99',\n",
       " 'REC9A',\n",
       " 'REC9B',\n",
       " 'REC9C',\n",
       " 'RECDV',\n",
       " 'RECECD',\n",
       " 'RECFG',\n",
       " 'RECG1',\n",
       " 'RECG2',\n",
       " 'RECGC',\n",
       " 'RECH0',\n",
       " 'RECH1',\n",
       " 'RECH10',\n",
       " 'RECH11',\n",
       " 'RECH2',\n",
       " 'RECH3',\n",
       " 'RECH4',\n",
       " 'RECH4A',\n",
       " 'RECH5',\n",
       " 'RECH5A',\n",
       " 'RECH5B',\n",
       " 'RECH5CS',\n",
       " 'RECH5S',\n",
       " 'RECH6',\n",
       " 'RECH6A',\n",
       " 'RECH6B',\n",
       " 'RECH6CS',\n",
       " 'RECH6S',\n",
       " 'RECH7',\n",
       " 'RECH7A',\n",
       " 'RECH7B',\n",
       " 'RECH7C',\n",
       " 'RECH7D',\n",
       " 'RECH8',\n",
       " 'RECH9',\n",
       " 'RECH9A',\n",
       " 'RECHA',\n",
       " 'RECHAA',\n",
       " 'RECHAC',\n",
       " 'RECHAN1',\n",
       " 'RECHAN2',\n",
       " 'RECHAN3',\n",
       " 'RECHB',\n",
       " 'RECHBB',\n",
       " 'RECHC',\n",
       " 'RECHCD',\n",
       " 'RECHCH',\n",
       " 'RECHCH1',\n",
       " 'RECHCH2',\n",
       " 'RECHCHL',\n",
       " 'RECHD',\n",
       " 'RECHDI',\n",
       " 'RECHDIS',\n",
       " 'RECHDP',\n",
       " 'RECHDP2',\n",
       " 'RECHEL',\n",
       " 'RECHEM',\n",
       " 'RECHFAC',\n",
       " 'RECHG1',\n",
       " 'RECHG2',\n",
       " 'RECHGS1',\n",
       " 'RECHGS2',\n",
       " 'RECHI',\n",
       " 'RECHII',\n",
       " 'RECHIL',\n",
       " 'RECHIV',\n",
       " 'RECHLB',\n",
       " 'RECHM1',\n",
       " 'RECHM2',\n",
       " 'RECHM3',\n",
       " 'RECHMA',\n",
       " 'RECHMB',\n",
       " 'RECHMC',\n",
       " 'RECHMC2',\n",
       " 'RECHMCS',\n",
       " 'RECHMG',\n",
       " 'RECHMH',\n",
       " 'RECHMH2',\n",
       " 'RECHML',\n",
       " 'RECHML2',\n",
       " 'RECHMLS',\n",
       " 'RECHMS',\n",
       " 'RECHMT',\n",
       " 'RECHMW',\n",
       " 'RECHOV',\n",
       " 'RECHPC',\n",
       " 'RECHS',\n",
       " 'RECHSA',\n",
       " 'RECHSK',\n",
       " 'RECHVC',\n",
       " 'RECHW',\n",
       " 'RECHYT',\n",
       " 'RECML',\n",
       " 'RECWS'}"
      ]
     },
     "execution_count": 74,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "db_helper._verified_tables"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## See what was modified\n",
    "\n",
    "For any tables that had columns added / widened, you **may** want to reload all data into that table. It depends on how the table has been updated in the past and what files you're running against: is it possible that some data files already in the DB were loaded without all necessary columns being present? If the DB has been kept up to date using this code, then it shouldn't be an issue. \n",
    "\n",
    "Otherwise you might want to set RELOAD_ALL_MODIFIED to True. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "set()"
      ]
     },
     "execution_count": 75,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "tables_changed = db_helper.list_modified_tables()\n",
    "tables_changed"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "RELOAD_ALL_MODIFIED = False"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Load the data!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for table_file in data_files:    \n",
    "    # check how many (if any) rows for this survey exist in this table\n",
    "    surveyid, _, _, _, table_name = TableDataHelper.parse_table_name(table_file)\n",
    "    if surveyid != '524':\n",
    "        continue\n",
    "    n_in_db = db_helper.get_db_survey_table_rowcount(surveyid, table_name)\n",
    "    print(os.path.basename(table_file) + \"... \", end=\"\")\n",
    "    if n_in_db == 0:\n",
    "        print(\"\\n    ....File needs loading completely\")\n",
    "        db_helper.load_table(table_file)\n",
    "        continue\n",
    "    try:\n",
    "        data = pd.read_csv(table_file)\n",
    "    except UnicodeDecodeError:\n",
    "        # You might need to keep an eye on this. So far, this is the only other \n",
    "        # encoding I've seen.\n",
    "        data = pd.read_csv(table_file, encoding='cp1252')\n",
    "    n_in_file = len(data)\n",
    "    if n_in_file > n_in_db:\n",
    "        print(\"\\n    ....File has more rows than db; drop and reload\")\n",
    "        db_helper.drop_and_reload(table_file)\n",
    "    elif RELOAD_ALL_MODIFIED and table_name in db_helper.list_modified_tables():\n",
    "        print(\"\\n    ....DB table had schema modified; drop and reload\")\n",
    "        db_helper.drop_and_reload(table_file)\n",
    "    \n",
    "    else:\n",
    "        print(\"... ok!\")\n",
    "    "
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.8.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}