Merge pull request #449 from DalgoT4D/create-orgtasks

Create org task

ddpui/api/data_api.py

@@ -0,0 +1,78 @@
+"""All the master data api cane found here"""
+
+from ninja import NinjaAPI
+from ninja.errors import HttpError
+from django.forms.models import model_to_dict
+
+from ninja.errors import ValidationError
+from ninja.responses import Response
+from pydantic.error_wrappers import ValidationError as PydanticValidationError
+
+
+from ddpui import auth
+from ddpui.models.tasks import Task
+from ddpui.utils.custom_logger import CustomLogger
+from ddpui.utils.constants import (
+    TASK_DBTRUN,
+    TASK_GITPULL,
+)
+from ddpui.ddpdbt import dbt_service
+
+dataapi = NinjaAPI(urls_namespace="master_data")
+# http://127.0.0.1:8000/api/docs
+
+
+logger = CustomLogger("ddpui")
+
+
+@dataapi.exception_handler(ValidationError)
+def ninja_validation_error_handler(request, exc):  # pylint: disable=unused-argument
+    """
+    Handle any ninja validation errors raised in the apis
+    These are raised during request payload validation
+    exc.errors is correct
+    """
+    return Response({"detail": exc.errors}, status=422)
+
+
+@dataapi.exception_handler(PydanticValidationError)
+def pydantic_validation_error_handler(
+    request, exc: PydanticValidationError
+):  # pylint: disable=unused-argument
+    """
+    Handle any pydantic errors raised in the apis
+    These are raised during response payload validation
+    exc.errors() is correct
+    """
+    print(exc)
+    return Response({"detail": exc.errors()}, status=500)
+
+
+@dataapi.exception_handler(Exception)
+def ninja_default_error_handler(
+    request, exc: Exception  # skipcq PYL-W0613
+):  # pylint: disable=unused-argument
+    """Handle any other exception raised in the apis"""
+    logger.info(exc)
+    return Response({"detail": "something went wrong"}, status=500)
+
+
+@dataapi.get("/tasks/", auth=auth.CanManagePipelines())
+def get_tasks(request):
+    """Fetch master list of tasks related to transformation"""
+    tasks = [
+        model_to_dict(task, exclude=["id"])
+        for task in Task.objects.filter(type__in=["dbt", "git"]).all()
+    ]
+    return tasks
+
+
+@dataapi.get("/tasks/{slug}/config/", auth=auth.CanManagePipelines())
+def get_task_config(request, slug):
+    """Get task config which details about the parameters that can be added/used while running it"""
+    task = Task.objects.filter(slug=slug).first()
+
+    if not task:
+        raise HttpError(404, "Task not found")
+
+    return dbt_service.task_config_params(task)

ddpui/ddpdbt/dbt_service.py

@@ -1,6 +1,7 @@
 import os
 import shutil
 from ddpui.models.org_user import Org
+from ddpui.models.tasks import Task
 from ddpui.models.org import OrgPrefectBlock
 from ddpui.ddpprefect import prefect_service
 from ddpui.ddpprefect import DBTCORE, SHELLOPERATION
@@ -35,3 +36,17 @@ def delete_dbt_workspace(org: Org):
             shellblock.delete()
 
     secretsmanager.delete_github_token(org)
+
+
+def task_config_params(task: Task):
+    """Return the config dictionary to setup parameters on this task"""
+
+    # dbt task config parameters
+    TASK_CONIF_PARAM = {
+        "dbt-deps": {"flags": ["upgrade"], "options": ["add-package"]},
+        "dbt-run": {"flags": ["full-refresh"], "options": ["select", "exclude"]},
+        "dbt-test": {"flags": [], "options": ["select", "exclude"]},
+        "dbt-seed": {"flags": [], "options": ["select"]},
+    }
+
+    return TASK_CONIF_PARAM[task.slug] if task.slug in TASK_CONIF_PARAM else None

ddpui/urls.py

@@ -12,12 +12,14 @@
 from ddpui.api.webhook_api import webhookapi
 from ddpui.api.superset_api import supersetapi
 from ddpui.api.pipeline_api import pipelineapi
+from ddpui.api.data_api import dataapi
 from ddpui.healthcheck import healthcheck
 
 urlpatterns = [
     # path("admin/", admin.site.urls), # Uncomment if you want to use django-admin app
     path("api/dashboard/", dashboardapi.urls),
     path("api/airbyte/", airbyteapi.urls),
+    path("api/data/", dataapi.urls),
     path("api/dbt/", dbtapi.urls),
     path("api/prefect/tasks/", orgtaskapi.urls),
     path("api/prefect/", pipelineapi.urls),