refactor

Author: toddbirchard

.gitignore

@@ -109,6 +109,8 @@ HackersAndSlackers-d2a47db89384.json
 HackersAndSlackers-4893023543f3.json
 HackersAndSlackers-59ed81beb2ea.json
 HackersAndSlackers-aec129ee8154.json
+hackersandslackers-204807-a78d7cf1d0d8.json
+*.json
 
 # .DS_Store
 .DS_Store

Pipfile.lock

@@ -1,11 +1,11 @@
-from src.sources.bigquery import bqc
-from src.sources.database import dbc
-from src.queries import queries
+from biquery_sql_etl.sources.bigquery import bqc
+from biquery_sql_etl.sources.sqldatabase import dbc
+from biquery_sql_etl.queries import sql_queries
 
 
 def main():
     """Move data between sources."""
-    for k, v in queries.items():
+    for k, v in sql_queries.items():
         fetched_rows = bqc.fetch_rows(v)
         inserted_rows = dbc.insert_rows(fetched_rows, k, replace=True)
         print(inserted_rows)

src/__init__.py biquery_sql_etl/__init__.pysrc/__init__.py renamed to biquery_sql_etl/__init__.py

@@ -0,0 +1,40 @@
+"""Read SQL queries from local files."""
+from os import listdir
+from os.path import isfile, join
+from config import local_sql_folder
+
+
+def get_local_sql_files():
+    """Fetch all SQL query files in folder."""
+    files = [local_sql_folder + '/' + f for f in listdir(local_sql_folder) if isfile(join(local_sql_folder, f)) if '.sql' in f]
+    return files
+
+
+def read_sql_queries():
+    """Read SQL query from .sql file."""
+    files = get_local_sql_files()
+    file_contents = []
+    for file in files:
+        fd = open(file, 'r')
+        query = fd.read()
+        file_contents.append(query)
+        fd.close()
+    return file_contents
+
+
+def create_query_dict():
+    """Create dictionary of queries for logging purposes."""
+    file_names = listdir(local_sql_folder)
+    file_contents = read_sql_queries()
+    table_names = get_table_name(file_names)
+    files = dict(zip(table_names, file_contents))
+    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()

biquery_sql_etl/queries.py

@@ -0,0 +1,2 @@
+from .bigquery import bqc
+from .sqldatabase import dbc

biquery_sql_etl/sources/__init__.py

@@ -1,7 +1,7 @@
-from config.bigquery_config import (gcp_credentials,
-                                    gcp_project,
-                                    bigquery_dataset,
-                                    bigquery_table)
+from config import (gcp_credentials,
+                    gcp_project,
+                    bigquery_dataset,
+                    bigquery_table)
 from .bigquery_client import BigQueryClient
 
 bqc = BigQueryClient(project=gcp_project,

src/sources/bigquery/__init__.py …ery_sql_etl/sources/bigquery/__init__.pysrc/sources/bigquery/__init__.py renamed to biquery_sql_etl/sources/bigquery/__init__.py src/sources/bigquery/__init__.py renamed to biquery_sql_etl/sources/bigquery/__init__.py

@@ -1,7 +1,7 @@
 """BigQuery Client."""
 from sqlalchemy.engine import create_engine
 from sqlalchemy import MetaData
-from src.sources.client import BaseClient
+from biquery_sql_etl.sources.client import BaseClient
 
 
 class BigQueryClient(BaseClient):

src/sources/bigquery/bigquery_client.py …_etl/sources/bigquery/bigquery_client.pysrc/sources/bigquery/bigquery_client.py renamed to biquery_sql_etl/sources/bigquery/bigquery_client.py src/sources/bigquery/bigquery_client.py renamed to biquery_sql_etl/sources/bigquery/bigquery_client.py

@@ -11,26 +11,27 @@
 class BaseClient:
 
     def __init__(self, engine=None, metadata=None, table=None):
-        # self.table = table
         self.engine = engine
         self.metadata = metadata
+        self.table = table
 
     def insert_rows(self, rows, table, replace=None):
         """Insert rows into table."""
         if replace:
             self.engine.execute(f'TRUNCATE TABLE {table}')
         table = Table(table, self.metadata, autoload=True)
         self.engine.execute(table.insert(), rows)
-        return self.__construct_response(rows)
+        return self.__construct_response(table, rows)
 
     def fetch_rows(self, query):
         """Fetch all rows via query."""
         rows = self.engine.execute(query).fetchall()
         return rows
 
-    def __construct_response(self, rows):
+    @staticmethod
+    def __construct_response(table, rows):
         """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 `{self.table}` with {len(columns)} columns: {column_names}'
+        return f'Inserted {num_rows} rows into `{table}` with {len(columns)} columns: {column_names}'

src/sources/client.py biquery_sql_etl/sources/client.pysrc/sources/client.py renamed to biquery_sql_etl/sources/client.py

@@ -1,9 +1,9 @@
-from config.sql_config import (database_username,
-                               database_password,
-                               database_host,
-                               database_port,
-                               database_table,
-                               database_name)
+from config import (database_username,
+                    database_password,
+                    database_host,
+                    database_port,
+                    database_table,
+                    database_name)
 from .database_client import Database
 
 dbc = Database('mysql',

src/sources/database/__init__.py …_sql_etl/sources/sqldatabase/__init__.pysrc/sources/database/__init__.py renamed to biquery_sql_etl/sources/sqldatabase/__init__.py src/sources/database/__init__.py renamed to biquery_sql_etl/sources/sqldatabase/__init__.py

@@ -1,30 +1,30 @@
 """SQL Database Client."""
 from sqlalchemy.engine import create_engine
 from sqlalchemy import MetaData
-from src.sources.client import BaseClient
+from biquery_sql_etl.sources.client import BaseClient
 
 
 class Database(BaseClient):
 
     def __init__(self,
-                 type,
+                 db_type,
                  username=None,
                  password=None,
                  host=None,
                  port=None,
                  table=None,
                  db_name=None):
         self.table = table
-        self.type = self.__conection_type(type)
+        self.type = self.__connection_type(db_type)
         self.engine = create_engine(f'{self.type}://{username}:{password}@{host}:{port}/{db_name}')
         super().__init__(engine=self.engine,
                          metadata=MetaData(bind=self.engine),
                          table=table)
 
     @staticmethod
-    def __conection_type(type):
-        if type.lower() == 'mysql':
+    def __connection_type(db_type):
+        if db_type.lower() == 'mysql':
             return 'mysql+pymysql'
-        if type.lower() == 'postgres':
+        if db_type.lower() == 'postgres':
             return 'postgresql+psycopg2'
         return None