Adding Pandas sample file

khalidmammadov · khalidmammadov · commit b7030667eeab · 2020-02-23T12:04:50.000Z
diff --git a/pd_sql/.gitignore b/pd_sql/.gitignore
@@ -0,0 +1,2 @@
+pyodbc/
+venv/
diff --git a/pd_sql/__init__.py b/pd_sql/__init__.py
diff --git a/pd_sql/__main__.py b/pd_sql/__main__.py
@@ -0,0 +1,68 @@
+import argparse
+from pathlib import Path
+from app import App
+
+
+def parse_arguments():
+    parser = argparse.ArgumentParser(description='Liability Cash Flow Generation Tool')
+    parser.add_argument('--log',
+                        action='store_true',
+                        help='Log messages')
+    parser.add_argument('--debug',
+                        action='store_true',
+                        help='Save files in various stages into csv files')
+    parser.add_argument('--debug_dir',
+                        required=False,
+                        help='Dir where to save CSV files to')
+    parser.add_argument('--dbpwd',
+                        required=False,
+                        help='Password to SQL Server (for Linux ODBC drivers)')
+    args = parser.parse_args()
+
+    return args.log, args.debug, args.debug_dir, args.dbpwd
+
+
+if __name__ == '__main__':
+    """
+    This small application demonstrates how it's to read from a CSV file and
+    save it into a database.
+    NOTE:
+    There is some overhead here with date type (datetime64) when read 
+    from a database which is relevant only in Linux and ok in Windows.  
+    
+    params: Supply password as a params when executed for Linux systems.
+            No need in windows as you can save the password in the ODBC 
+            connection settings 
+    """
+    # Parse arguments
+    logging_enabled, debug_enabled, debug_dir, dbpwd = parse_arguments()
+
+    print(f'Logging:{logging_enabled}')
+    print(f'Debug:{debug_enabled}')
+    print(f'Debug dir:{debug_dir}')
+
+    # Initialise a global app
+    # this will hold the state and components of the app
+    app = App(log_flag=logging_enabled,
+              debug_flag=debug_enabled,
+              debug_dir=debug_dir,
+              dbpwd=dbpwd)
+
+    # Main implementation
+    csv_path = Path('~/docker/test/exchange-rates/data')
+    daily_csv_file = 'daily.csv'
+
+    # Read file to a DataFrame
+    csv_df = app.get_csv_data(csv_path.joinpath(daily_csv_file))
+    csv_df = csv_df.astype({'Date': 'datetime64'})
+    csv_df = csv_df.rename(columns={'Date': 'RateDate'})
+    csv_df = csv_df.fillna(0)
+    # Check the data
+    app.debug_df('Rates from CSV:', csv_df)
+
+    # Save Rates Into DB
+    app.save_rates_into_db(csv_df)
+
+
+
+
diff --git a/pd_sql/app.py b/pd_sql/app.py
@@ -0,0 +1,133 @@
+import logging
+import pandas as pd
+from pathlib import Path
+from time import time
+from functools import wraps
+
+from db.connection import new_connection
+from db.lib import insert_rates, update_rates
+from db.Findb.Daily_rates import daily_rates_tbl, get_all_rates
+from util.config import load_config, get_config
+
+
+def timing(f):
+    @wraps(f)
+    def wrap(*args, **kw):
+        logger = args[0].logger
+        ts = time()
+        logger.info('*' * 50)
+        logger.info('Executing: {}()...'.format(f.__name__.upper()))
+        logger.info('*' * 50)
+        result = f(*args, **kw)
+        te = time()
+        logger.info('*' * 50)
+        logger.info('{}() took: {:.0f}min {:.2f}sec '.format(f.__name__.upper(), (te-ts)/60, te-ts))
+        logger.info('*' * 50)
+        return result
+    return wrap
+
+
+class App(object):
+
+    def __init__(self, log_flag=None, debug_flag=None, debug_dir=None, dbpwd=None):
+
+        self.logger = App.set_up_logger(log_flag, debug_flag)
+        self.debug_enabled = debug_flag
+        self.debug_dir = None
+
+        self.config = load_config('conf.ini')
+        dsn_name = self.get_config('DSN', 'Name')
+        self.db_connection = new_connection(dsn_name, dbpwd)
+
+    @staticmethod
+    def set_up_logger(log_flag, debug_flag):
+
+        # Set logging level
+        logging_level = logging.ERROR
+        if log_flag:
+            logging_level = logging.INFO
+
+        if debug_flag:
+            logging_level = logging.DEBUG
+
+        # Setup logger
+        logging.basicConfig(level=logging_level,
+                            format='%(asctime)-15s %(name)-5s %(levelname)-8s %(message)s')
+
+        logger = logging.getLogger(__name__)
+        logger.setLevel(logging_level)
+
+        return logger
+
+    def set_up_debug_dir(self):
+        dir_ = None
+        if self.debug_enabled:
+            dir_ = Path(self.debug_dir)
+            if not dir_.exists():
+                raise Exception(f"Dir: {self.debug_dir} does not exists!")
+        return dir_
+
+    def debug_dump_df(self, name, df):
+
+        # Check if folder setup
+        if not self.debug_dir:
+            self.debug_dir = App.set_up_debug_dir()
+
+        if self.debug_enabled:
+            file = self.debug_dir.joinpath(name).with_suffix('.csv')
+            self.debug(f'Dumping into file: {file}')
+            df.to_csv(file)
+
+    def debug(self, msg):
+        self.logger.debug(msg)
+
+    def log(self, msg):
+        self.logger.info(msg)
+
+    def debug_df(self, title, df):
+        self.debug('\n' + title + ' - (head) \n' + df.head().to_string())
+
+    @timing
+    def get_config(self, section, name):
+        return get_config(self.config, section, name)
+
+    @staticmethod
+    def get_csv_data(file):
+        return pd.read_csv(file)
+
+    @timing
+    def save_rates_into_db(self, rates_df):
+        # Get rates from DB
+        rates_db_df = get_all_rates(self.db_connection)
+        rates_db_df = rates_db_df.astype({'RateDate': 'datetime64'})
+        # Left Outer Join so we check for existence
+        joined = pd.merge(rates_df,
+                          rates_db_df,
+                          left_on=['RateDate', 'Country'],
+                          right_on=['RateDate', 'Country'],
+                          how='left',
+                          suffixes=['', '_table'])
+
+        # Separate Inserts and updates
+        cond = joined['Value_table'].isnull()
+        inserts = joined[cond]
+        updates = joined[~cond]
+
+        # Check
+        self.debug_df('Inserts:', inserts)
+        self.debug_df('Updates:', updates)
+
+        if inserts.index.size > 0:
+            insert_rates(self.db_connection,
+                         daily_rates_tbl,
+                         inserts)
+
+        if updates.index.size > 0:
+            update_rates(self.db_connection,
+                         daily_rates_tbl,
+                         updates)
+
+    def __del__(self):
+        if self.db_connection:
+            self.db_connection.close()
+            del self.db_connection
diff --git a/pd_sql/conf.ini b/pd_sql/conf.ini
@@ -0,0 +1,4 @@
+
+
+[DSN]
+Name=MSSQLServerDatabase
diff --git a/pd_sql/db/Findb/Daily_rates.py b/pd_sql/db/Findb/Daily_rates.py
@@ -0,0 +1,22 @@
+from sqlalchemy import DECIMAL, String, Date, MetaData
+from db.lib import LabeledColumn, LabeledTable, make_select_list, table_reader
+
+db_schema = 'Findb.dbo'
+daily_rates_tbl = LabeledTable(
+                     'daily_rates', MetaData()
+                     , LabeledColumn('RateDate', Date, select_exp="format(RateDate, 'yyyyMMdd')")
+                     , LabeledColumn('Country', String)
+                     , LabeledColumn('Value', DECIMAL)
+                     , schema=db_schema
+                     )
+
+
+def get_all_rates(connection):
+    select_list = make_select_list(daily_rates_tbl.select_columns_exp_dict)
+    df = table_reader(
+                      connection
+                      , columns=select_list
+                      , table=f'{db_schema}.{daily_rates_tbl.name}'
+                      )
+    return df
+
diff --git a/pd_sql/db/Findb/__init__.py b/pd_sql/db/Findb/__init__.py
diff --git a/pd_sql/db/__init__.py b/pd_sql/db/__init__.py
diff --git a/pd_sql/db/connection.py b/pd_sql/db/connection.py
@@ -0,0 +1,12 @@
+import sqlalchemy as sqla
+import platform
+
+
+def new_connection(dsn_name, dbpwd=None):
+    print('Creating new connection')
+
+    if platform.system() == 'Linux':
+        return sqla.create_engine(f"mssql+pyodbc://sa:{dbpwd}@{dsn_name}", echo=True).connect()
+
+    return sqla.create_engine(f"mssql+pyodbc://{dsn_name}", echo=True).connect()
+
diff --git a/pd_sql/db/lib.py b/pd_sql/db/lib.py
@@ -0,0 +1,74 @@
+import pandas as pd
+from sqlalchemy import Column, Table, and_
+from lib.converter import make_upsert_list, make_select_list
+
+
+class Db:
+    def __init__(self, name, schema):
+        self.name = name
+        self.schema = schema
+        self.db_schema = '.'.join([name, schema])
+
+
+class LabeledColumn(Column):
+
+    def __init__(self, name, *args, label=None, select_exp=None, **kwargs):
+        Column.__init__(self, name, *args, **kwargs)
+        self.label = label if label else name
+        self.select_exp = select_exp
+
+
+class LabeledTable(Table):
+
+    def __init__(self, *args, **kwargs):
+        Table.__init__(self, *args, **kwargs)
+
+    @property
+    def labels(self):
+        return [col.label for col in self.columns]
+
+    @property
+    def column_label_dict(self):
+        return {col.name: col.label for col in self.columns}
+
+    @property
+    def select_columns_exp_dict(self):
+        return {col.select_exp
+                if col.select_exp
+                else col.name: col.label
+                for col in self.columns}
+
+    @property
+    def select_columns_dict(self):
+        return {col.name: col.label
+                for col in self.columns}
+
+
+def table_reader(connection, table, columns, condition=''):
+    cond = '' if condition == '' else f'where {condition}'
+    sql = 'select {} from {} {}'.format(','.join(columns), table, cond)
+
+    return pd.read_sql(sql, connection)
+
+
+def insert_rates(connection, table, inserts):
+    # Make dictionary of parameters for the inserts
+    ins_params = make_upsert_list(inserts)
+    connection.execute(
+        table.insert(),
+        ins_params)
+
+
+def update_rates(connection, table, updates):
+    # Make dictionary of parameters for the updates
+    upd_params = make_upsert_list(updates)
+    # Execute statement (will autocommit)
+    # Generate single updates
+    for uld in upd_params:
+        connection.execute(
+            table.update().
+                where(
+                    and_(
+                        table.c.RateDate == uld['RateDate'],
+                        table.c.Country == uld['Country'])),
+            upd_params)
diff --git a/pd_sql/util/__init__.py b/pd_sql/util/__init__.py
diff --git a/pd_sql/util/config.py b/pd_sql/util/config.py
@@ -0,0 +1,13 @@
+import configparser
+
+
+def load_config(config_file):
+    config = configparser.ConfigParser()
+    config.read(config_file)
+    return config
+
+
+def get_config(config, section, name):
+    return config[section][name]
+
+

-Original file line number
+Diff line change
@@ @@ -0,0 +1,4 @@ @@
++
++
 +[DSN]
 +Name=MSSQLServerDatabase