refactor(dag:FB_POST_INSIGHTS_V1): improve typing and simplify logic

Author: Lee-W

dags/ods/fb_post_insights/dag.py

@@ -30,16 +30,14 @@
 def FB_POST_INSIGHTS_V1():
     @task
     def CREATE_TABLE_IF_NEEDED():
-        client = bigquery.Client(project=os.getenv("BIGQUERY_PROJECT"))
-        post_sql = """
+        create_post_table_sql = """
         CREATE TABLE IF NOT EXISTS `pycontw-225217.ods.ods_pycontw_fb_posts` (
             id STRING,
             created_at TIMESTAMP,
             message STRING
         )
         """
-        client.query(post_sql)
-        insights_sql = """
+        create_insights_table_sql = """
         CREATE TABLE IF NOT EXISTS `pycontw-225217.ods.ods_pycontw_fb_posts_insights` (
             post_id STRING,
             query_time TIMESTAMP,
@@ -48,24 +46,29 @@ def CREATE_TABLE_IF_NEEDED():
             share INTEGER
         )
         """
-        client.query(insights_sql)
+
+        bg_project = os.getenv("BIGQUERY_PROJECT")
+        client = bigquery.Client(project=bg_project)
+        client.query(create_post_table_sql)
+        client.query(create_insights_table_sql)
 
     @task
     def SAVE_FB_POSTS_AND_INSIGHTS():
         posts = request_posts_data()
 
         last_post = query_last_post()
-        if last_post is None:
-            new_posts = posts
-        else:
-            new_posts = [
+        new_posts = (
+            [
                 post
                 for post in posts
                 if datetime.strptime(
                     post["created_time"], "%Y-%m-%dT%H:%M:%S%z"
                 ).timestamp()
                 > last_post["created_at"].timestamp()
             ]
+            if last_post is not None
+            else posts
+        )
 
         if not dump_posts_to_bigquery(
             [

dags/ods/fb_post_insights/udfs.py

@@ -113,9 +113,12 @@ def dump_posts_to_bigquery(posts: list[dict]) -> bool:
     client = bigquery.Client(project=os.getenv("BIGQUERY_PROJECT"))
     job_config = bigquery.LoadJobConfig(
         schema=[
-            bigquery.SchemaField("id", "STRING", mode="REQUIRED"),
-            bigquery.SchemaField("created_at", "TIMESTAMP", mode="REQUIRED"),
-            bigquery.SchemaField("message", "STRING", mode="REQUIRED"),
+            bigquery.SchemaField(field_name, field_type, mode="REQUIRED")
+            for field_name, field_type in [
+                ("id", "STRING"),
+                ("created_at", "TIMESTAMP"),
+                ("message", "STRING"),
+            ]
         ],
         write_disposition="WRITE_APPEND",
     )
@@ -128,6 +131,7 @@ def dump_posts_to_bigquery(posts: list[dict]) -> bool:
         job.result()
         return True
     except Exception as e:
+        # TODO: catch with more specific exception
         logger.error(f"Failed to dump posts to BigQuery: {e}", exc_info=True)
         return False
 

dags/ods/ig_post_insights/udfs.py

@@ -27,15 +27,13 @@
 
 
 def create_table_if_needed() -> None:
-    client = bigquery.Client(project=os.getenv("BIGQUERY_PROJECT"))
     post_sql = """
     CREATE TABLE IF NOT EXISTS `pycontw-225217.ods.ods_pycontw_ig_posts` (
         id STRING,
         created_at TIMESTAMP,
         message STRING
     )
     """
-    client.query(post_sql)
     insights_sql = """
     CREATE TABLE IF NOT EXISTS `pycontw-225217.ods.ods_pycontw_ig_posts_insights` (
         post_id STRING,
@@ -47,6 +45,9 @@ def create_table_if_needed() -> None:
         views INTEGER
     )
     """
+
+    client = bigquery.Client(project=os.getenv("BIGQUERY_PROJECT"))
+    client.query(post_sql)
     client.query(insights_sql)
 
 
@@ -122,7 +123,6 @@ def request_posts_data() -> list[dict]:
     media_list = response.json()["data"]
 
     media_insight_list = []
-
     for media in media_list:
         media_insight_url = f"https://graph.facebook.com/v20.0/{media['id']}"
         querystring = {

dags/ods/linkedin_post_insights/udfs.py

@@ -10,15 +10,13 @@
 
 
 def create_table_if_needed() -> None:
-    client = bigquery.Client(project=os.getenv("BIGQUERY_PROJECT"))
     post_sql = """
     CREATE TABLE IF NOT EXISTS `pycontw-225217.ods.ods_pycontw_linkedin_posts` (
         id STRING,
         created_at TIMESTAMP,
         message STRING
     )
     """
-    client.query(post_sql)
     insights_sql = """
     CREATE TABLE IF NOT EXISTS `pycontw-225217.ods.ods_pycontw_linkedin_posts_insights` (
         post_id STRING,
@@ -30,6 +28,8 @@ def create_table_if_needed() -> None:
         views INTEGER
     )
     """
+    client = bigquery.Client(project=os.getenv("BIGQUERY_PROJECT"))
+    client.query(post_sql)
     client.query(insights_sql)
 
     # Example output from the Rapid API, not all fields will exists for a specific post