refactor(dag:*post_insights): extract bq_client as cached_property

Author: Lee-W

dags/utils/posts_insights/base.py

@@ -1,6 +1,7 @@
 import logging
 import os
 from abc import ABC, abstractmethod
+from functools import cached_property
 
 from google.cloud import bigquery
 
@@ -35,15 +36,18 @@ class BasePostsInsightsParser(ABC):
     CREATE_POSTS_TABLE_SQL: str = ""
     CREATE_INSIGHTS_TABLE_SQL: str = ""
 
+    @cached_property
+    def bq_client(self) -> bigquery.Client:
+        return bigquery.Client(project=os.getenv("BIGQUERY_PROJECT", ""))
+
     def create_tables_if_not_exists(self) -> None:
         if not self.CREATE_POSTS_TABLE_SQL and not self.CREATE_INSIGHTS_TABLE_SQL:
             raise ValueError(
                 "Both the SQLs to create table for posts and insights must be set"
             )
 
-        client = bigquery.Client(project=os.getenv("BIGQUERY_PROJECT", ""))
         for sql in [self.CREATE_POSTS_TABLE_SQL, self.CREATE_INSIGHTS_TABLE_SQL]:
-            client.query(sql)
+            self.bq_client.query(sql)
 
     def save_posts_and_insights(self) -> None:
         posts = self._request_posts_data()
@@ -65,7 +69,6 @@ def _request_posts_data(self) -> list[dict]: ...
     def _filter_new_posts(self, posts: list[dict], last_post: dict) -> list[dict]: ...
 
     def _query_last_post(self) -> dict | None:
-        client = bigquery.Client(project=os.getenv("BIGQUERY_PROJECT"))
         sql = f"""
         SELECT
             created_at
@@ -75,7 +78,7 @@ def _query_last_post(self) -> dict | None:
             created_at DESC
         LIMIT 1
         """
-        result = client.query(sql)
+        result = self.bq_client.query(sql)
         data = list(result)
         return data[0] if data else None
 
@@ -87,7 +90,6 @@ def _dump_posts_to_bigquery(self, posts: list[dict]) -> None:
             logger.info("No posts to dump!")
             return
 
-        client = bigquery.Client(project=os.getenv("BIGQUERY_PROJECT"))
         job_config = bigquery.LoadJobConfig(
             schema=[
                 bigquery.SchemaField("id", "STRING", mode="REQUIRED"),
@@ -97,7 +99,7 @@ def _dump_posts_to_bigquery(self, posts: list[dict]) -> None:
             write_disposition="WRITE_APPEND",
         )
         try:
-            job = client.load_table_from_json(
+            job = self.bq_client.load_table_from_json(
                 posts,
                 f"pycontw-225217.ods.{self.POST_TABLE_NAME}",
                 job_config=job_config,
@@ -115,7 +117,6 @@ def _dump_posts_insights_to_bigquery(self, posts: list[dict]) -> None:
             logger.info("No post insights to dump!")
             return
 
-        client = bigquery.Client(project=os.getenv("BIGQUERY_PROJECT"))
         job_config = bigquery.LoadJobConfig(
             schema=[
                 bigquery.SchemaField("post_id", "STRING", mode="REQUIRED"),
@@ -129,7 +130,7 @@ def _dump_posts_insights_to_bigquery(self, posts: list[dict]) -> None:
             write_disposition="WRITE_APPEND",
         )
         try:
-            job = client.load_table_from_json(
+            job = self.bq_client.load_table_from_json(
                 posts,
                 f"pycontw-225217.ods.{self.INSIGHT_TABLE_NAME}",
                 job_config=job_config,

dags/utils/posts_insights/facebook.py

@@ -1,5 +1,4 @@
 import logging
-import os
 from datetime import datetime
 
 import requests
@@ -68,7 +67,6 @@ def _dump_posts_to_bigquery(self, posts: list[dict]) -> None:
             logger.info("No posts to dump!")
             return
 
-        client = bigquery.Client(project=os.getenv("BIGQUERY_PROJECT"))
         job_config = bigquery.LoadJobConfig(
             schema=[
                 bigquery.SchemaField(field_name, field_type, mode="REQUIRED")
@@ -81,15 +79,15 @@ def _dump_posts_to_bigquery(self, posts: list[dict]) -> None:
             write_disposition="WRITE_APPEND",
         )
         try:
-            job = client.load_table_from_json(
+            job = self.bq_client.load_table_from_json(
                 posts,
                 "pycontw-225217.ods.ods_pycontw_fb_posts",
                 job_config=job_config,
             )
             job.result()
         except Exception:
             logger.exception("Failed to dump posts to BigQuery: ")
-            raise RuntimeError("Failed to dump posts insights to BigQuery")
+            raise RuntimeError("Failed to dump posts to BigQuery")
 
     def _process_posts_insights(self, posts: list[dict]) -> list[dict]:
         return [
@@ -108,7 +106,6 @@ def _dump_posts_insights_to_bigquery(self, posts: list[dict]) -> None:
             logger.info("No post insights to dump!")
             return
 
-        client = bigquery.Client(project=os.getenv("BIGQUERY_PROJECT"))
         job_config = bigquery.LoadJobConfig(
             schema=[
                 bigquery.SchemaField("post_id", "STRING", mode="REQUIRED"),
@@ -120,15 +117,15 @@ def _dump_posts_insights_to_bigquery(self, posts: list[dict]) -> None:
             write_disposition="WRITE_APPEND",
         )
         try:
-            job = client.load_table_from_json(
+            job = self.bq_client.load_table_from_json(
                 posts,
                 "pycontw-225217.ods.ods_pycontw_fb_posts_insights",
                 job_config=job_config,
             )
             job.result()
         except Exception:
             logger.exception("Failed to dump posts insights to BigQuery: ")
-            raise RuntimeError("Failed to dump posts to BigQuery")
+            raise RuntimeError("Failed to dump posts insights to BigQuery")
 
 
 def convert_fb_time(time_string: str) -> str: