refactor: Update add env in table name if env is empty string

airflow/dags/gmaps/d_gmaps_dim_places.py

@@ -1,7 +1,7 @@
 from datetime import datetime, timedelta
 
 from google.cloud import bigquery
-from utils.common import load_config
+from utils.common import load_config, table_name_with_env
 from utils.gcp import query_bq
 
 from airflow.decorators import dag, task
@@ -10,7 +10,7 @@
 BQ_ODS_DATASET = config["gcp"]["bigquery"]["ods_dataset"]
 BQ_DIM_DATASET = config["gcp"]["bigquery"]["dim_dataset"]
 ODS_TABLE_NAME = "ods-" + config["gcp"]["table"]["gmaps-places"]
-DIM_TABLE_NAME = "dim-gmaps-places-" + config["env"]
+DIM_TABLE_NAME = table_name_with_env("dim-gmaps-places", config["env"])
 
 BQ_CLIENT = bigquery.Client()
 

airflow/dags/gmaps/d_gmaps_dim_time.py

@@ -1,7 +1,7 @@
 from datetime import datetime, timedelta
 
 from google.cloud import bigquery
-from utils.common import load_config
+from utils.common import load_config, table_name_with_env
 from utils.gcp import query_bq
 
 from airflow.decorators import dag, task
@@ -10,7 +10,7 @@
 BQ_ODS_DATASET = config["gcp"]["bigquery"]["ods_dataset"]
 BQ_DIM_DATASET = config["gcp"]["bigquery"]["dim_dataset"]
 ODS_TABLE_NAME = "ods-" + config["gcp"]["table"]["gmaps-reviews"]
-DIM_TABLE_NAME = "dim-time-" + config["env"]
+DIM_TABLE_NAME = table_name_with_env("dim-time", config["env"])
 
 BQ_CLIENT = bigquery.Client()
 

airflow/dags/gmaps/d_gmaps_dim_users.py

@@ -1,7 +1,7 @@
 from datetime import datetime, timedelta
 
 from google.cloud import bigquery
-from utils.common import load_config
+from utils.common import load_config, table_name_with_env
 from utils.gcp import query_bq
 
 from airflow.decorators import dag, task
@@ -10,7 +10,7 @@
 BQ_ODS_DATASET = config["gcp"]["bigquery"]["ods_dataset"]
 BQ_DIM_DATASET = config["gcp"]["bigquery"]["dim_dataset"]
 ODS_TABLE_NAME = "ods-" + config["gcp"]["table"]["gmaps-reviews"]
-DIM_TABLE_NAME = "dim-gmaps-users-" + config["env"]
+DIM_TABLE_NAME = table_name_with_env("dim-gmaps-users", config["env"])
 
 BQ_CLIENT = bigquery.Client()
 

airflow/dags/gmaps/d_gmaps_fact_reviews.py

@@ -2,7 +2,7 @@
 
 import pandas as pd
 from google.cloud import bigquery
-from utils.common import load_config
+from utils.common import load_config, table_name_with_env
 from utils.gcp import query_bq
 
 from airflow.decorators import dag, task
@@ -11,7 +11,7 @@
 BQ_ODS_DATASET = config["gcp"]["bigquery"]["ods_dataset"]
 BQ_FACT_DATASET = config["gcp"]["bigquery"]["fact_dataset"]
 ODS_TABLE_NAME = "ods-" + config["gcp"]["table"]["gmaps-reviews"]
-FACT_TABLE_NAME = "fact-gmaps-reviews-" + config["env"]
+FACT_TABLE_NAME = table_name_with_env("fact-gmaps-reviews", config["env"])
 
 BQ_CLIENT = bigquery.Client()
 

airflow/dags/gmaps/d_gmaps_mart_review_trends.py

@@ -1,7 +1,7 @@
 from datetime import datetime, timedelta
 
 from google.cloud import bigquery
-from utils.common import load_config
+from utils.common import load_config, table_name_with_env
 from utils.gcp import query_bq
 
 from airflow.decorators import dag, task
@@ -13,7 +13,10 @@
 BQ_FACT_DATASET = config["gcp"]["bigquery"]["fact_dataset"]
 BQ_MART_DATASET = config["gcp"]["bigquery"]["mart_dataset"]
 ODS_TABLE_NAME = "ods-" + config["gcp"]["table"]["gmaps-reviews"]
-MART_TABLE_NAME = "mart-reviews-trends-" + config["env"]
+fact_reviews_table = table_name_with_env("fact-gmaps-reviews", config["env"])
+dim_places_table = table_name_with_env("dim-gmaps-places", config["env"])
+dim_time_table = table_name_with_env("dim-time", config["env"])
+MART_TABLE_NAME = table_name_with_env("mart-reviews-trends", config["env"])
 
 BQ_CLIENT = bigquery.Client()
 
@@ -81,12 +84,12 @@ def l_mart_review_trends():
           COUNT(r.`review_id`) AS `total_reviews`,
           ROUND(AVG(r.`rating`), 2) AS `avg_rating`,
         FROM
-          `{BQ_FACT_DATASET}`.`{"fact-gmaps-reviews-" + config["env"]}` r
+          `{BQ_FACT_DATASET}`.`{fact_reviews_table}` r
         JOIN
-          `{BQ_DIM_DATASET}`.`{"dim-gmaps-places-" + config["env"]}` p
+          `{BQ_DIM_DATASET}`.`{dim_places_table}` p
           ON r.`place_name` = p.`place_name`
         JOIN
-          `{BQ_DIM_DATASET}`.`{"dim-time-" + config["env"]}` t
+          `{BQ_DIM_DATASET}`.`{dim_time_table}` t
           ON r.`published_at` = t.`date`
         GROUP BY
           p.`city`,

airflow/utils/common.py

@@ -63,7 +63,9 @@ def add_env_suffix(config: dict) -> dict:
         if "table" in config["gcp"]:
             for key, value in config["gcp"]["table"].items():
                 if isinstance(value, str) and not value.startswith("ENV_"):
-                    config["gcp"]["table"][key] = f"{value}-{env_suffix}"
+                    config["gcp"]["table"][key] = (
+                        f"{value}-{env_suffix}" if env_suffix else value
+                    )
 
         # Add suffix to blob prefixes and update all paths
         if "blob" in config["gcp"]:
@@ -73,7 +75,11 @@ def add_env_suffix(config: dict) -> dict:
                     if isinstance(
                         original_prefix, str
                     ) and not original_prefix.startswith("ENV_"):
-                        new_prefix = f"{original_prefix}-{env_suffix}"
+                        new_prefix = (
+                            f"{original_prefix}-{env_suffix}"
+                            if env_suffix
+                            else original_prefix
+                        )
                         blob_info["prefix"] = new_prefix
                         # Update all paths under this blob
                         for path_key, path_value in blob_info.items():
@@ -104,3 +110,7 @@ def load_config() -> dict:
     config = add_env_suffix(config)
 
     return config
+
+
+def table_name_with_env(base_name, env):
+    return f"{base_name}-{env}" if env else base_name