Import google spreadsheet into bigquery

dags/google_spreadsheet_import_pipeline.py

@@ -0,0 +1,47 @@
+import os
+import logging
+from datetime import timedelta
+from airflow import DAG
+
+from data_pipeline.utils.dags.data_pipeline_dag_utils import (
+    get_default_args,
+    create_python_task,
+)
+from data_pipeline.spreadsheet_data.google_spreadsheet_config import (
+    MultiCsvSheet
+)
+from data_pipeline.spreadsheet_data.google_spreadsheet_etl import (
+    etl_google_spreadsheet
+)
+
+LOGGER = logging.getLogger(__name__)
+
+DEPLOYMENT_ENV_ENV_NAME = "DEPLOYMENT_ENV"
+DEFAULT_DEPLOYMENT_ENV_VALUE = "ci"
+
+DAG_ID = "Google_Spreadsheet_Data_Pipeline"
+G_SPREADSHEET_DAG = DAG(
+    dag_id=DAG_ID,
+    default_args=get_default_args(),
+    schedule_interval=None,
+    dagrun_timeout=timedelta(minutes=60),
+)
+
+
+def get_env_var_or_use_default(env_var_name, default_value):
+    return os.getenv(env_var_name, default_value)
+
+
+def google_spreadsheet_data_etl(**kwargs):
+    data_config_dict = kwargs["dag_run"].conf
+    dep_env = get_env_var_or_use_default(
+        DEPLOYMENT_ENV_ENV_NAME, DEFAULT_DEPLOYMENT_ENV_VALUE
+    )
+    data_config = MultiCsvSheet(data_config_dict, dep_env)
+    etl_google_spreadsheet(data_config)
+
+
+G_SPREADSHEET_ETL_TASK = create_python_task(
+    G_SPREADSHEET_DAG, "google_spreadsheet_data_etl",
+    google_spreadsheet_data_etl,
+)

dags/google_spreadsheet_pipeline_controller.py

@@ -0,0 +1,57 @@
+import os
+
+from airflow import DAG
+
+from data_pipeline.spreadsheet_data.google_spreadsheet_config import (
+    MultiSpreadsheetConfig,
+)
+from data_pipeline.spreadsheet_data.google_spreadsheet_etl import (
+    get_yaml_file_as_dict
+)
+from data_pipeline.utils.dags.data_pipeline_dag_utils import (
+    get_default_args,
+    simple_trigger_dag,
+    create_python_task
+)
+
+GOOGLE_SPREADSHEET_SCHEDULE_INTERVAL_ENV_NAME = (
+    "GOOGLE_SPREADSHEET_SCHEDULE_INTERVAL"
+)
+SPREADSHEET_CONFIG_FILE_PATH_ENV_NAME = (
+    "SPREADSHEET_CONFIG_FILE_PATH"
+)
+
+DEPLOYMENT_ENV_ENV_NAME = "DEPLOYMENT_ENV"
+DEFAULT_DEPLOYMENT_ENV_VALUE = "ci"
+
+TARGET_DAG = "Google_Spreadsheet_Data_Pipeline"
+
+
+def get_env_var_or_use_default(env_var_name, default_value=None):
+    return os.getenv(env_var_name, default_value)
+
+
+# pylint: disable=unused-argument
+def trigger_dag(**kwargs):
+    conf_file_path = get_env_var_or_use_default(
+        SPREADSHEET_CONFIG_FILE_PATH_ENV_NAME, ""
+    )
+    data_config_dict = get_yaml_file_as_dict(conf_file_path)
+
+    data_config = MultiSpreadsheetConfig(data_config_dict,)
+    for _, spreadsheet_config in data_config.spreadsheets_config.items():
+        simple_trigger_dag(dag_id=TARGET_DAG, conf=spreadsheet_config)
+
+
+SPREADSHEET_CONTROLLER_DAG = DAG(
+    dag_id="Google_Spreadsheet_Import_Pipeline_Controller",
+    default_args=get_default_args(),
+    schedule_interval=get_env_var_or_use_default(
+        GOOGLE_SPREADSHEET_SCHEDULE_INTERVAL_ENV_NAME
+    ),
+)
+
+TRIGGER_SPEADSHEET_ETL_DAG_TASK = create_python_task(
+    SPREADSHEET_CONTROLLER_DAG, "trigger_google_spreadsheet_etl_dag",
+    trigger_dag,
+)

data_pipeline/spreadsheet_data/google_spreadsheet_config.py

@@ -0,0 +1,92 @@
+class MultiSpreadsheetConfig:
+    def __init__(self,
+                 multi_spreadsheet_config: dict,
+                 ):
+        self.gcp_project = multi_spreadsheet_config.get("gcpProjectName")
+        self.import_timestamp_field_name = multi_spreadsheet_config.get(
+            "importedTimestampFieldName"
+        )
+        self.spreadsheets_config = {
+            spreadsheet.get("spreadsheetId"): extend_spreadsheet_config_dict(
+                spreadsheet,
+                self.gcp_project,
+                self.import_timestamp_field_name,
+            )
+            for spreadsheet in multi_spreadsheet_config.get("spreadsheets")
+        }
+
+
+def extend_spreadsheet_config_dict(
+        spreadsheet_config_dict,
+        gcp_project: str,
+        imported_timestamp_field_name: str,
+):
+    spreadsheet_config_dict["gcpProjectName"] = gcp_project
+    spreadsheet_config_dict[
+        "importedTimestampFieldName"
+    ] = imported_timestamp_field_name
+
+    return spreadsheet_config_dict
+
+
+class MultiCsvSheet:
+    def __init__(self, multi_sheet_config: dict,
+                 deployment_env: str,
+                 ):
+        self.spreadsheet_id = multi_sheet_config.get("spreadsheetId")
+        self.import_timestamp_field_name = multi_sheet_config.get(
+            "importedTimestampFieldName"
+        )
+        self.gcp_project = multi_sheet_config.get("gcpProjectName")
+        self.sheets_config = {
+            sheet.get("sheetName"): CsvSheetConfig(
+                sheet,
+                self.spreadsheet_id,
+                self.gcp_project,
+                self.import_timestamp_field_name,
+                deployment_env,
+            )
+            for sheet in multi_sheet_config.get("sheets")
+        }
+
+
+# pylint: disable=too-many-instance-attributes,too-many-arguments,
+# pylint: disable=simplifiable-if-expression
+class CsvSheetConfig:
+    def __init__(
+            self,
+            csv_sheet_config: dict,
+            spreadsheet_id: str,
+            gcp_project: str,
+            imported_timestamp_field_name: str,
+            deployment_env: str,
+            environment_placeholder: str = "{ENV}"
+    ):
+        self.gcp_project = gcp_project
+        self.import_timestamp_field_name = imported_timestamp_field_name
+        self.spreadsheet_id = spreadsheet_id
+        self.header_line_index = csv_sheet_config.get("headerLineIndex")
+        self.data_values_start_line_index = csv_sheet_config.get(
+            "dataValuesStartLineIndex"
+        )
+        self.sheet_name = csv_sheet_config.get("sheetName")
+        self.sheet_range = csv_sheet_config.get("sheetRange")
+        self.table_name = csv_sheet_config.get("tableName")
+        self.dataset_name = csv_sheet_config.get(
+            "datasetName"
+        ).replace(environment_placeholder, deployment_env)
+        self.table_write_append = (
+            True
+            if csv_sheet_config.get("tableWriteAppend", "").lower() == "true"
+            else False
+        )
+        self.metadata = {
+            record.get("metadataSchemaFieldName"):
+                record.get("metadataLineIndex")
+            for record in csv_sheet_config.get("metadata", [])
+        }
+        self.fixed_sheet_metadata = {
+            record.get("metadataSchemaFieldName"):
+                record.get("fixedSheetValue")
+            for record in csv_sheet_config.get("fixedSheetMetadata", [])
+        }