Merge pull request #465 from DalgoT4D/451-api-to-fetch-sources-in-a-schema

451 api to fetch sources in a schema

Author: fatchat

ddpui/api/transform_api.py

@@ -18,7 +18,7 @@
 from ddpui.utils.custom_logger import CustomLogger
 
 from ddpui.schemas.org_task_schema import DbtProjectSchema
-from ddpui.schemas.dbt_workflow_schema import CreateDbtModelPayload
+from ddpui.schemas.dbt_workflow_schema import CreateDbtModelPayload, SyncSourcesSchema
 
 from ddpui.core import dbtautomation_service
 
@@ -121,6 +121,32 @@ def delete_dbt_project(request, project_name: str):
     return {"message": f"Project {project_name} deleted successfully"}
 
 
+@transformapi.post("/sync_sources/", auth=auth.CanManagePipelines())
+def sync_sources(request, payload: SyncSourcesSchema):
+    """
+    Sync sources from a given schema.
+    """
+    orguser: OrgUser = request.orguser
+    org = orguser.org
+
+    org_warehouse = OrgWarehouse.objects.filter(org=org).first()
+    if not org_warehouse:
+        raise HttpError(status_code=404, detail="Please set up your warehouse first")
+
+    orgdbt = OrgDbt.objects.filter(org=org, gitrepo_url=None).first()
+    if not orgdbt:
+        raise HttpError(status_code=404, detail="DBT workspace not set up")
+
+    sources_file_path, error = dbtautomation_service.sync_sources_to_dbt(
+        payload.schema_name, payload.source_name, org, org_warehouse
+    )
+
+    if error:
+        raise HttpError(status_code=422, detail=error)
+
+    return {"sources_file_path": str(sources_file_path)}
+
+
 ########################## Models #############################################
 
 

ddpui/core/dbtautomation_service.py

@@ -1,19 +1,23 @@
+import os
 from pathlib import Path
-from ddpui.models.org import OrgWarehouse, OrgDbt
-from ddpui.utils import secretsmanager
-from dbt_automation.utils.warehouseclient import get_client
 
-# operations
-from dbt_automation.operations.flattenjson import flattenjson
 from dbt_automation.operations.arithmetic import arithmetic
 from dbt_automation.operations.castdatatypes import cast_datatypes
 from dbt_automation.operations.coalescecolumns import coalesce_columns
 from dbt_automation.operations.concatcolumns import concat_columns
 from dbt_automation.operations.droprenamecolumns import drop_columns, rename_columns
 from dbt_automation.operations.flattenairbyte import flatten_operation
+
+# operations
+from dbt_automation.operations.flattenjson import flattenjson
 from dbt_automation.operations.mergetables import union_tables
 from dbt_automation.operations.regexextraction import regex_extraction
+from dbt_automation.operations.syncsources import sync_sources
+from dbt_automation.utils.warehouseclient import get_client
+from dbt_automation.utils.dbtproject import dbtProject
 
+from ddpui.models.org import OrgDbt, OrgWarehouse
+from ddpui.utils import secretsmanager
 
 OPERATIONS_DICT = {
     "flatten": flatten_operation,
@@ -52,3 +56,20 @@ def create_dbt_model_in_project(
     )
 
     return str(sql_file_path), None
+
+
+def sync_sources_to_dbt(
+    schema_name: str, source_name: str, org: str, org_warehouse: str
+):
+    """
+    Sync sources from a given schema to dbt.
+    """
+    warehouse_client = _get_wclient(org_warehouse)
+
+    sources_file_path = sync_sources(
+        config={"source_schema": schema_name, "source_name": source_name},
+        warehouse=warehouse_client,
+        dbtproject=dbtProject(Path(os.getenv("CLIENTDBT_ROOT")) / org.slug / "dbtrepo"),
+    )
+
+    return str(sources_file_path), None

ddpui/schemas/dbt_workflow_schema.py

@@ -1,4 +1,4 @@
-from ninja import Schema, Field
+from ninja import Field, Schema
 
 
 class CreateDbtModelPayload(Schema):
@@ -10,3 +10,12 @@ class CreateDbtModelPayload(Schema):
     display_name: str
     config: dict
     op_type: str
+
+
+class SyncSourcesSchema(Schema):
+    """
+    schema to sync sources from the schema
+    """
+
+    schema_name: str
+    source_name: str