database flavor agnostic, bidirectional ETL

toddbirchard · toddbirchard · commit 173a656e7672 · 2019-11-13T01:26:21.000-05:00
diff --git a/.gitignore b/.gitignore
@@ -103,11 +103,21 @@ venv.bak/
 # mypy
 .mypy_cache/
 
-
+# Google Cloud Creds
 HackersAndSlackers-d2a47db89384.json
 HackersAndSlackers-d2a47db89384.json
 HackersAndSlackers-4893023543f3.json
 HackersAndSlackers-59ed81beb2ea.json
 HackersAndSlackers-aec129ee8154.json
-.env
+
+# .DS_Store
 .DS_Store
+
+# Pycharm
+.idea
+.idea/.gitignore
+.idea/bigquery-to-mysql.iml
+.idea/inspectionProfiles/
+.idea/misc.xml
+.idea/modules.xml
+.idea/vcs.xml
diff --git a/Pipfile b/Pipfile
@@ -6,10 +6,11 @@ verify_ssl = true
 [dev-packages]
 
 [packages]
-google-cloud-bigquery = "*"
-sqlalchemy = "*"
-pybigquery = "*"
-pymysql = "*"
+Google-Cloud-BigQuery = "*"
+SQLAlchemy = "*"
+PyBigQuery = "*"
+PyMySQL = "*"
+Psycopg2-Binary = "*"
 
 [requires]
 python_version = "3.7"
diff --git a/Pipfile.lock b/Pipfile.lock
diff --git a/README.md b/README.md
@@ -5,9 +5,10 @@
 ![PyBigQuery](https://img.shields.io/badge/PyBigQuery-v0.4.11-blue.svg?logo=Google&longCache=true&logoColor=white&colorB=5e81ac&style=flat-square&colorA=4c566a)
 ![SQLAlchemy](https://img.shields.io/badge/SQLAlchemy-v1.3.1-red.svg?longCache=true&style=flat-square&logo=scala&logoColor=white&colorA=4c566a&colorB=bf616a)
 ![PyMySQL](https://img.shields.io/badge/PyMySQL-v0.9.3-red.svg?longCache=true&style=flat-square&logo=mysql&logoColor=white&colorA=4c566a&colorB=bf616a)
+![Psycopg2-Binary](https://img.shields.io/badge/Psycopg2--Binary-v2.8.4-red.svg?longCache=true&style=flat-square&logo=postgresql&logoColor=white&colorA=4c566a&colorB=bf616a)
 ![GitHub Last Commit](https://img.shields.io/github/last-commit/google/skia.svg?style=flat-square&colorA=4c566a&colorB=a3be8c)
 [![GitHub Issues](https://img.shields.io/github/issues/toddbirchard/bigquery-to-sql.svg?style=flat-square&colorA=4c566a&colorB=ebcb8b)](https://github.com/hackersandslackers/bigquery-python-tutorial/issues)
 [![GitHub Stars](https://img.shields.io/github/stars/toddbirchard/bigquery-to-sql.svg?style=flat-square&colorB=ebcb8b&colorA=4c566a)](https://github.com/hackersandslackers/bigquery-python-tutorial/stargazers)
 [![GitHub Forks](https://img.shields.io/github/forks/toddbirchard/bigquery-to-sql.svg?style=flat-square&colorA=4c566a&colorB=ebcb8b)](https://github.com/hackersandslackers/bigquery-python-tutorial/network)
 
-Lightweight ETL script to migrate data from BigQuery to SQL.
+Lightweight ETL script to migrate data from BigQuery to SQL database, and vice versa.
diff --git a/config.py b/config.py
@@ -11,14 +11,27 @@ class Config:
     # Google BigQuery
     bigquery_dataset = environ.get('GCP_BIGQUERY_DATASET')
     bigquery_table = environ.get('GCP_BIGQUERY_TABLE')
-    bigquery_uri = f'bigquery://{gcp_project}/{bigquery_dataset}'
-    bigquery_query = environ.get('GCP_BIGQUERY_QUERY')
+    bigquery_query = 'queries/analytics.sql'
 
-    # MySQL
-    mysql_username = environ.get('MYSQL_USERNAME')
-    mysql_password = environ.get('MYSQL_PASSWORD')
-    mysql_host = environ.get('MYSQL_HOST')
-    mysql_port = environ.get('MYSQL_PORT')
-    mysql_db = environ.get('MYSQL_DATABASE_NAME')
-    mysql_table = environ.get('MYSQL_TABLE_NAME')
-    mysql_uri = f'mysql+pymysql://{mysql_username}:{mysql_password}@{mysql_host}/{mysql_db}'
+    # SQL
+    database_username = environ.get('DATABASE_USERNAME')
+    database_password = environ.get('DATABASE_PASSWORD')
+    database_host = environ.get('DATABASE_HOST')
+    database_port = environ.get('DATABASE_PORT')
+    database_db = environ.get('DATABASE_DATABASE_NAME')
+    database_table = environ.get('DATABASE_TABLE_NAME')
+
+    @classmethod
+    def get_bigquery_config(cls):
+        bigquery_uri = f'bigquery://{cls.gcp_project}/{cls.bigquery_dataset}'
+        return bigquery_uri, cls.gcp_credentials
+
+    @classmethod
+    def get_database_config(cls, type=None):
+        database_flavor = None
+        if type.lower() == 'mysql':
+            database_flavor = 'mysql+pymysql'
+        if type.lower() == 'postgres':
+            database_flavor = 'postgresql+psycopg2'
+        database_uri = f'{database_flavor}://{cls.database_username}:{cls.database_password}@{cls.database_host}:{cls.database_port}/{cls.database_db}'
+        return database_uri, cls.database_table
diff --git a/queries/analytics.sql b/queries/analytics.sql
@@ -0,0 +1,16 @@
+SELECT
+  REPLACE(REPLACE(REPLACE(title, ' | Hackers and Slackers', ''), ' | Hackers And Slackers', ''), '- Hackers and Slackers', '') as title,
+  url,
+  REPLACE(REPLACE(url, 'https://hackersandslackers.com/', ''), '/' , '') as slug,
+  COUNT(title) AS views
+FROM
+  hackersgatsbyprod.pages
+WHERE
+  timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 day)
+GROUP BY
+  title,
+  url
+ORDER BY
+  COUNT(title) DESC
+LIMIT
+  100;
diff --git a/requirements.txt b/requirements.txt
@@ -10,6 +10,7 @@ google-resumable-media==0.4.1
 googleapis-common-protos==1.6.0
 idna==2.8
 protobuf==3.10.0
+psycopg2-binary==2.8.4
 pyasn1==0.4.7
 pyasn1-modules==0.2.7
 pybigquery==0.4.11
@@ -18,5 +19,5 @@ pytz==2019.3
 requests==2.22.0
 rsa==4.0
 six==1.13.0
-SQLAlchemy==1.3.10
-urllib3==1.25.6
+SQLAlchemy==1.3.11
+urllib3==1.25.7
diff --git a/src/__init__.py b/src/__init__.py
@@ -1,15 +1,27 @@
-import logging
-from .bigquery import BigQueryClient
-from .database import MySQLClient
 from config import Config
-
-logging.basicConfig(format='%(asctime)s \n%(message)s',
-                    level=logging.INFO)
+from src.bigquery import BigQueryClient
+from src.database import SQLClient
+from src.util.read import get_query_from_file
 
 
 def main():
-    bigquery = BigQueryClient(Config)
-    mysql = MySQLClient(Config)
-    bq_rows = bigquery.fetch_bigquery_rows(Config.bigquery_query)
-    updated = mysql.insert_rows(bq_rows)
-    logging.info(f'Top posts this week: {updated}')
+    bigquery = init_bigquery()
+    database = init_database()
+    query = get_query_from_file(Config.bigquery_query)
+    rows = bigquery.fetch_rows(query)
+    updated = database.insert_rows(rows, replace=True)
+    print(updated)
+
+
+def init_bigquery():
+    """Initiate BigQuery Client."""
+    uri, creds = Config.get_bigquery_config()
+    bigquery = BigQueryClient(uri=uri, creds=creds)
+    return bigquery
+
+
+def init_database():
+    """Initiate SQL database Client."""
+    uri, table = Config.get_database_config(type='mysql')
+    bigquery = SQLClient(uri=uri, table=table)
+    return bigquery
diff --git a/src/bigquery.py b/src/bigquery.py
@@ -1,18 +1,30 @@
-"""BigQuery data source."""
+"""BigQuery Client."""
+import logging
 from sqlalchemy.engine import create_engine
-from sqlalchemy import MetaData
+
+
+logging.basicConfig(filename='logs/bigquery.log',
+                    format='%(asctime)s %(message)s')
+logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)
 
 
 class BigQueryClient:
 
-    def __init__(self, Config):
-        self.metadata = MetaData()
-        self.uri = Config.bigquery_uri
-        self.credentials = Config.gcp_credentials
+    def __init__(self, uri=None, table=None, creds=None):
+        self.uri = uri
+        self.table_name = table
+        self.creds = creds
         self.engine = create_engine(self.uri,
-                                    credentials_path=self.credentials)
+                                    credentials_path=self.creds)
+
+    def insert_rows(self, rows, replace=None):
+        """Insert rows into SQL table."""
+        if replace:
+            self.engine.execute(f'TRUNCATE TABLE {self.table_name}')
+        self.engine.execute(self.table.insert(), rows)
+        return [row.items()[0][1] for row in rows]
 
-    def fetch_bigquery_rows(self, query):
+    def fetch_rows(self, query):
         """Fetch rows from BigQuery via query."""
         results = self.engine.execute(query).fetchall()
         return results
diff --git a/src/database.py b/src/database.py
@@ -1,24 +1,32 @@
-"""MySQL data destination."""
+"""SQL Database Client."""
+import logging
 from sqlalchemy.engine import create_engine
-from sqlalchemy import MetaData, Table, Column, Integer, String
+from sqlalchemy import MetaData, Table
 
+logging.basicConfig(filename='logs/sql.log',
+                    format='%(asctime)s %(message)s')
+logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)
 
-class MySQLClient:
 
-    def __init__(self, Config):
-        self.mysql_uri = Config.mysql_uri
-        self.engine = create_engine(self.mysql_uri)
-        self.metadata = MetaData()
-        self.table_name = Config.mysql_table
-        self.table = Table(Config.mysql_table, self.metadata,
-                           Column('id', Integer),
-                           Column('title', String),
-                           Column('url', String),
-                           Column('slug', String),
-                           Column('views', Integer))
+class SQLClient:
 
-    def insert_rows(self, rows):
-        """Insert rows into MySQL table."""
-        self.engine.execute(f'TRUNCATE TABLE {self.table_name}')
+    def __init__(self, uri=None, table=None):
+        self.uri = uri
+        self.table_name = table
+        self.engine = create_engine(self.uri)
+        self.metadata = MetaData(bind=self.engine)
+        self.table = Table(table,
+                           self.metadata,
+                           autoload=True)
+
+    def insert_rows(self, rows, replace=None):
+        """Insert rows into SQL table."""
+        if replace:
+            self.engine.execute(f'TRUNCATE TABLE {self.table_name}')
         self.engine.execute(self.table.insert(), rows)
         return [row.items()[0][1] for row in rows]
+
+    def fetch_rows(self, query):
+        """Fetch rows from SQL table via query."""
+        results = self.engine.execute(query).fetchall()
+        return results
diff --git a/src/util/read.py b/src/util/read.py
@@ -0,0 +1,9 @@
+"""Read SQL queries from local files."""
+
+
+def get_query_from_file(path):
+    """Read contents of file."""
+    file = open(path, 'r')
+    query = file.read()
+    file.close()
+    return query
