Ingest the file into the database table.
(excel_file, db_name, table_name, db_type="pgsql", schema=None)
| 107 | |
| 108 | |
| 109 | def ingest(excel_file, db_name, table_name, db_type="pgsql", schema=None): |
| 110 | """Ingest the file into the database table.""" |
| 111 | logging.info( |
| 112 | f"file = {excel_file}, db = {db_name}, table = {table_name}, db type = {db_type}" |
| 113 | ) |
| 114 | |
| 115 | # Create database engine |
| 116 | db = db_provider.get(db_type) |
| 117 | engine = db.get_engine() |
| 118 | |
| 119 | # Inspect the target table schema |
| 120 | inspector = inspect(engine) |
| 121 | dtypes = {} |
| 122 | for column in inspector.get_columns(table_name, schema=schema): |
| 123 | dtypes[column["name"]] = column["type"] |
| 124 | logging.info(dtypes) |
| 125 | |
| 126 | # Load the excel into database |
| 127 | df = pd.read_excel(excel_file, engine="openpyxl") |
| 128 | df.to_sql( |
| 129 | table_name, engine, if_exists="append", chunksize=500, index=False, dtype=dtypes |
| 130 | ) |
| 131 | |
| 132 | # TODO - Validation |
| 133 | print(f"\nTotal records in {excel_file} - {len(df)}") |
| 134 | for c in df.columns: |
| 135 | print(f"{c} - {df[c].nunique()}") |