452 api to fetch models sources from a local dbt project

ddpui/api/transform_api.py

@@ -121,7 +121,7 @@ def delete_dbt_project(request, project_name: str):
     return {"message": f"Project {project_name} deleted successfully"}
 
 
-@transformapi.post("/sync_sources/", auth=auth.CanManagePipelines())
+@transformapi.post("/dbt_project/sync_sources/", auth=auth.CanManagePipelines())
 def sync_sources(request, payload: SyncSourcesSchema):
     """
     Sync sources from a given schema.
@@ -147,10 +147,10 @@ def sync_sources(request, payload: SyncSourcesSchema):
     return {"sources_file_path": str(sources_file_path)}
 
 
-########################## Models #############################################
+########################## Models & Sources #############################################
 
 
-@transformapi.post("/model/", auth=auth.CanManagePipelines())
+@transformapi.post("/dbt_project/model/", auth=auth.CanManagePipelines())
 def post_dbt_model(request, payload: CreateDbtModelPayload):
     """
     Create a model on local disk and save configuration to django db
@@ -173,6 +173,7 @@ def post_dbt_model(request, payload: CreateDbtModelPayload):
         raise HttpError(422, "model output name must be unique")
 
     payload.config["output_name"] = output_name
+    payload.config["dest_schema"] = payload.dest_schema
 
     sql_path, error = dbtautomation_service.create_dbt_model_in_project(
         orgdbt, org_warehouse, payload.op_type, payload.config
@@ -184,8 +185,42 @@ def post_dbt_model(request, payload: CreateDbtModelPayload):
         orgdbt=orgdbt,
         name=output_name,
         display_name=payload.display_name,
+        schema=payload.dest_schema,
         sql_path=sql_path,
         config=payload.config,
     )
 
     return model_to_dict(orgdbt_model, exclude=["orgdbt", "id"])
+
+
+@transformapi.get("/dbt_project/sources_models/", auth=auth.CanManagePipelines())
+def get_input_sources_and_models(request, schema_name: str = None):
+    """
+    Fetches all sources and models in a dbt project
+    """
+    orguser: OrgUser = request.orguser
+    org = orguser.org
+
+    org_warehouse = OrgWarehouse.objects.filter(org=org).first()
+    if not org_warehouse:
+        raise HttpError(404, "please setup your warehouse first")
+
+    # make sure the orgdbt here is the one we create locally
+    orgdbt = OrgDbt.objects.filter(org=org, gitrepo_url=None).first()
+    if not orgdbt:
+        raise HttpError(404, "dbt workspace not setup")
+
+    sources = dbtautomation_service.read_dbt_sources_in_project(orgdbt)
+
+    models = []
+    for orgdbt_model in OrgDbtModel.objects.filter(orgdbt=orgdbt).all():
+        models.append(
+            {
+                "source_name": None,
+                "input_name": orgdbt_model.name,
+                "input_type": "model",
+                "schema": orgdbt_model.schema,
+            }
+        )
+
+    return [ref for ref in sources + models if ref["schema"] == schema_name]

ddpui/core/dbtautomation_service.py

@@ -15,6 +15,7 @@
 from dbt_automation.operations.syncsources import sync_sources
 from dbt_automation.utils.warehouseclient import get_client
 from dbt_automation.utils.dbtproject import dbtProject
+from dbt_automation.utils.dbtsources import read_sources
 
 from ddpui.models.org import OrgDbt, OrgWarehouse
 from ddpui.utils import secretsmanager
@@ -73,3 +74,9 @@ def sync_sources_to_dbt(
     )
 
     return str(sources_file_path), None
+
+
+def read_dbt_sources_in_project(orgdbt: OrgDbt):
+    """Read the sources from .yml files in the dbt project"""
+
+    return read_sources(Path(orgdbt.project_dir) / "dbtrepo")

ddpui/migrations/0046_orgdbtmodel_schema.py

@@ -0,0 +1,17 @@
+# Generated by Django 4.1.7 on 2024-02-14 10:43
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("ddpui", "0045_orgdbtmodel"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="orgdbtmodel",
+            name="schema",
+            field=models.CharField(max_length=100, null=True),
+        ),
+    ]

ddpui/models/dbt_workflow.py

@@ -13,6 +13,7 @@ class OrgDbtModel(models.Model):
     orgdbt = models.ForeignKey(OrgDbt, on_delete=models.CASCADE)
     name = models.CharField(max_length=100)
     display_name = models.CharField(max_length=100)
+    schema = models.CharField(max_length=100, null=True)
     sql_path = models.CharField(max_length=200, null=True)
     config = models.JSONField(null=True)
 

ddpui/schemas/dbt_workflow_schema.py

@@ -8,6 +8,7 @@ class CreateDbtModelPayload(Schema):
 
     name: str
     display_name: str
+    dest_schema: str
     config: dict
     op_type: str