reafactor

Author: toddbirchard

README.md

@@ -27,7 +27,6 @@ $ pipenv update
 $ python3 main.py
 ```
 
-<<<<<<< HEAD
 Installation via [Poetry](https://python-poetry.org/):
 
 ```shell
@@ -37,8 +36,6 @@ $ poetry install
 $ poetry run
 ```
 
-=======
->>>>>>> 5ec1fda5600620aa22b8d53ad87226fe689b5151
 Alternatively, try installing via `setup.py`:
 
 ```shell

biquery_sql_etl/__init__.py

@@ -1,15 +1,18 @@
-from biquery_sql_etl.sources.bigquery import bqc
-from biquery_sql_etl.sources.sqldatabase import dbc
+from biquery_sql_etl.engines import bigquery_engine, rdbms_engine
 from biquery_sql_etl.queries import sql_queries
+from biquery_sql_etl.client import DataClient
 from loguru import logger
+from config import bigquery_table
 
 
 logger.add('logs/queries.log', format="{time} {message}", level="INFO")
 
 
 def main():
     """Move data between sources."""
-    for k, v in sql_queries.items():
-        fetched_rows = bqc.fetch_rows(v)
-        inserted_rows = dbc.insert_rows(fetched_rows, k, replace=True)
+    bqc = DataClient(bigquery_engine)
+    dbc = DataClient(rdbms_engine)
+    for table_name, query in sql_queries.items():
+        fetched_rows = bqc.fetch_rows(query, table=bigquery_table)
+        inserted_rows = dbc.insert_rows(fetched_rows, table_name, replace=True)
         logger.info(inserted_rows)

biquery_sql_etl/client.py

@@ -0,0 +1,37 @@
+"""Base Data Client."""
+from sqlalchemy import MetaData, Table
+
+
+class DataClient:
+
+    def __init__(self, engine):
+        self.engine = engine
+        self.metadata = MetaData(bind=self.engine)
+        self.table_name = None
+
+    @property
+    def table(self):
+        if self.table_name:
+            return Table(self.table_name, self.metadata, autoload=True)
+        return None
+
+    def insert_rows(self, rows, table=None, replace=None):
+        """Insert rows into table."""
+        if replace:
+            self.engine.execute(f'TRUNCATE TABLE {table}')
+        self.table_name = table
+        self.engine.execute(self.table.insert(), rows)
+        return self.construct_response(rows, table)
+
+    def fetch_rows(self, query, table=None):
+        """Fetch all rows via query."""
+        rows = self.engine.execute(query).fetchall()
+        return rows
+
+    @staticmethod
+    def construct_response(rows, table):
+        """Summarize results of an executed query."""
+        columns = rows[0].keys()
+        column_names = ", ".join(columns)
+        num_rows = len(rows)
+        return f'Inserted {num_rows} rows into `{table}` with {len(columns)} columns: {column_names}'

biquery_sql_etl/engines/__init__.py

@@ -0,0 +1,2 @@
+from .bigquery import bigquery_engine
+from .rdbms import rdbms_engine

biquery_sql_etl/engines/bigquery.py

@@ -0,0 +1,11 @@
+from sqlalchemy.engine import create_engine
+from config import (gcp_credentials,
+                    bigquery_uri)
+
+
+def create_bigquery_engine(uri, creds):
+    engine = create_engine(uri, credentials_path=creds)
+    return engine
+
+
+bigquery_engine = create_bigquery_engine(bigquery_uri, gcp_credentials)

biquery_sql_etl/engines/rdbms.py

@@ -0,0 +1,10 @@
+"""SQL Database Client."""
+from sqlalchemy.engine import create_engine
+from config import rdbms_uri
+
+
+def create_rdbms_engine(uri):
+    return create_engine(uri)
+
+
+rdbms_engine = create_rdbms_engine(rdbms_uri)

biquery_sql_etl/queries.py

@@ -22,6 +22,12 @@ def read_sql_queries():
     return file_contents
 
 
+def get_table_name(file_names):
+    """Assume that table names of queries are [FILE_NAME]_stats."""
+    tables = [f"{file.split('.')[0]}_stats" for file in file_names]
+    return tables
+
+
 def create_query_dict():
     """Create dictionary of queries for logging purposes."""
     file_names = listdir(local_sql_folder)
@@ -31,10 +37,4 @@ def create_query_dict():
     return files
 
 
-def get_table_name(file_names):
-    """Assume that table names of queries are [FILE_NAME]_stats."""
-    tables = [f"{file.split('.')[0]}_stats" for file in file_names]
-    return tables
-
-
 sql_queries = create_query_dict()