project_demo_2/Load_Environmental_Data.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "383a01ef",
   "metadata": {},
   "source": [
    "## Loading data from file to database"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f1558573",
   "metadata": {},
   "source": [
    "#### Import all libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "4c37ee2b",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyodbc\n",
    "import traceback\n",
    "from email.mime.text import MIMEText\n",
    "from email.mime.multipart import MIMEMultipart\n",
    "import smtplib\n",
    "import pandas as pd\n",
    "import glob\n",
    "import datetime\n",
    "import sqlalchemy as sa\n",
    "import shutil\n",
    "import warnings\n",
    "warnings.filterwarnings('ignore')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4e4e3788",
   "metadata": {},
   "source": [
    "#### Check if file exists and then load all files availale on a set path"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "66169c6b",
   "metadata": {},
   "outputs": [],
   "source": [
    "files = glob.glob(\"C:/Users/sayans/Documents/SharedFolderLocation_Inbound/Environmental_Data/*.csv\")\n",
    "archive_dir = \"C:/Users/sayans/Documents/SharedFolderLocation_Inbound/Environmental_Data/Archive/\"\n",
    "main_dataframe_raw = pd.DataFrame()\n",
    "no_files = 0\n",
    "if len(files)==0:\n",
    "    no_files=1\n",
    "    \n",
    "if no_files==0:\n",
    "    for i in range(0,len(files)):\n",
    "        data = pd.read_csv(files[i], dtype='str')\n",
    "        main_dataframe_raw = pd.concat([main_dataframe_raw,data],axis=0)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "415f810f",
   "metadata": {},
   "source": [
    "#### Perform all necessary data transformations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "60556bf8",
   "metadata": {},
   "outputs": [],
   "source": [
    "if no_files==0:\n",
    "    main_dataframe_stg1 = main_dataframe_raw[['Date UTC', 'Time UTC', 'Date', \\\n",
    "                                              'Time','Temperature','Relative Humidity','Dew Point']].copy()\n",
    "    main_dataframe_stg1['Time_copy'] = main_dataframe_stg1['Time'].apply(lambda x: x[:-4])\n",
    "    main_dataframe_stg1.loc[:,'Date_copy'] = pd.to_datetime(main_dataframe_stg1.loc[:,'Date']).dt.date\n",
    "    main_dataframe_stg1.loc[:,'Time_copy'] = pd.to_datetime(main_dataframe_stg1.loc[:,'Time']).dt.time\n",
    "\n",
    "    main_dataframe_stg1['Timestamp'] = main_dataframe_stg1.loc[:,['Date_copy','Time_copy']].apply(lambda x: \\\n",
    "                                                     datetime.datetime.combine(x['Date_copy'],x['Time_copy']), axis=1)\n",
    "\n",
    "    main_dataframe_stg1.drop(['Date_copy','Time_copy'], axis=1,inplace=True)\n",
    "\n",
    "    del main_dataframe_raw\n",
    "\n",
    "    #renaming columns as per available in DataBase\n",
    "    main_dataframe_stg1.rename(columns={'Date UTC':'Date_UTC', 'Time UTC':'Time_UTC','Date':'Date_EST','Time':'Time_EST',\\\n",
    "                                        'Temperature':'Temperature_fahrenheit','Dew Point':'Dewpoint_fahrenheit',\\\n",
    "                                        'Relative Humidity':'ReltiveHumidity_rh'\n",
    "                                       }, inplace=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b832083a",
   "metadata": {},
   "source": [
    "#### Connect to the database and insert into stage"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "e7a46a17",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "NoneType: None\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#set up engine\n",
    "setup = sa.create_engine(f'mssql+pyodbc://@ServerName/Database?driver=ODBC+Driver+17+for+SQL+Server')\n",
    "error = None\n",
    "\n",
    "try:\n",
    "    if no_files==0:\n",
    "        #setup connection\n",
    "        cnxn = setup.raw_connection()\n",
    "        cursor = cnxn.cursor()\n",
    "\n",
    "        #truncate the stage table\n",
    "        cursor.execute(\"TRUNCATE TABLE [dbo].[Environmental_Table_STG]\") \n",
    "        cnxn.commit()\n",
    "\n",
    "        #load data into stage table\n",
    "        main_dataframe_stg1[['Date_UTC', 'Time_UTC', 'Date_EST', 'Time_EST','Timestamp','Temperature_fahrenheit',\\\n",
    "                             'Dewpoint_fahrenheit','ReltiveHumidity_rh']].to_sql('Environmental_Monitoring_Index_STG',\\\n",
    "                                                                            setup,if_exists='append',index=False,schema='dbo') \n",
    "        #load data into main table\n",
    "        cursor.execute(\"EXEC [dbo].[load_Environmental_Table]\")\n",
    "\n",
    "        #read return data from SP\n",
    "        row = cursor.fetchone()\n",
    "        insert_row_count = row[0]\n",
    "        update_row_count = row[1]\n",
    "        load_time = row[2]\n",
    "\n",
    "        cnxn.commit()\n",
    "        \n",
    "        #move file to archive\n",
    "        for i in files:\n",
    "            fname_split = i.replace('C:/Users/sayans/Documents/SharedFolderLocation_Inbound/Environmental_Data\\\\','')\n",
    "            fname_split = fname_split.split('.')\n",
    "            fname_fxd = fname_split[0]+'_'+pd.Timestamp.now().strftime('%Y_%m_%d_%H_%M')+'.'+fname_split[1]\n",
    "            destpath = archive_dir+\"/\"+fname_fxd\n",
    "            shutil.move(i, destpath)\n",
    "            \n",
    "        #close all connections\n",
    "        cursor.close()\n",
    "        cnxn.close()\n",
    "        del cursor\n",
    "        del cnxn\n",
    "    \n",
    "except Exception as e:\n",
    "    error = e\n",
    "    cursor.close()\n",
    "    cnxn.close()\n",
    "    del cursor\n",
    "    del cnxn\n",
    "\n",
    "    \n",
    "finally:  \n",
    "    #build email\n",
    "    # we will built the message using the email library and send using smtplib\n",
    "    msg = MIMEMultipart()\n",
    "    msg['Subject'] = \"Automated Environmental Data Table Load Notification\"  # set email subject\n",
    "    \n",
    "    # we will send via outlook, first we initialise connection to mail server\n",
    "    smtp = smtplib.SMTP('smtp-mail.outlook.com', '587')\n",
    "    smtp.ehlo()  # say hello to the server\n",
    "    smtp.starttls()  # we will communicate using TLS encryption\n",
    "    \n",
    "    # login to outlook server, using generic email and password\n",
    "    smtp.login('donotreply@server.com', 'abcd1234')\n",
    "    \n",
    "    if (('cursor' in globals()) or ('cnxn' in globals())):\n",
    "        cursor.close()\n",
    "        cnxn.close()\n",
    "        del cursor\n",
    "        del cnxn\n",
    "    \n",
    "    if no_files==1:\n",
    "        # write an email message\n",
    "        txt = (f\"No Files are available to Proccess and Load into [dbo].[Environmental_Table_PRD] table\")\n",
    "        \n",
    "        msg.attach(MIMEText(txt))  # add text contents\n",
    "        \n",
    "        # send email\n",
    "        smtp.sendmail('donotreply@server.com', 'sayanswar@server.com', msg.as_string())\n",
    "    \n",
    "    elif(error==None):\n",
    "        # write an email message\n",
    "        txt = (f\"Load of [dbo].[Environmental_Table_PRD] Table is completed.\\nTotal {insert_row_count} rows were inserted and {update_row_count} rows were updated at {load_time}\")\n",
    "        \n",
    "        msg.attach(MIMEText(txt))  # add text contents\n",
    "        \n",
    "        # send email\n",
    "        smtp.sendmail('donotreply@server.com', 'sayanswar@server.com', msg.as_string())\n",
    "        \n",
    "    else:\n",
    "        txt = (f\"There was an error while executing the load script. \\nThe error is:-> {error}\")\n",
    "        \n",
    "        msg.attach(MIMEText(txt))  # add text contents\n",
    "        \n",
    "        # send email\n",
    "        smtp.sendmail('donotreply@server.com', 'sayanswar@server.com', msg.as_string())\n",
    "        \n",
    "    # finally, disconnect from the mail server\n",
    "    smtp.quit()\n",
    "    \n",
    "print(traceback.format_exc())\n",
    "#input(\"Press return to exit\")"
   ]
  }
 ],
 "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.10.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}