Ddp organization pipeline flow deployments prefect

.env.example

@@ -0,0 +1,3 @@
+# CONFIG
+DBT_VENV=/path/to/dbt/virtual/env
+PREFECT_WORK_QUEUE=ddp

.gitignore

@@ -1,12 +1,18 @@
 logs
 prefect-env
 dbt-env
-.env
+.*env
+*venv
 dbt/
 
+*.pyc
+
+.*yaml
+.*yml
+
 .DS_store
 
 # prefect artifacts
 .prefectignore
 
-*.json
+config.json

README.md

@@ -1 +1,15 @@
 # ddp_prefect_starter
+
+- Setup a virtual environment using ``` python3 -m venv <name> ```
+- Activate the virtual env ``` source <name>/bin/activate ```
+- Install the dependencies from requirements.txt ``` pip3 install -r requirements.txt ```
+- If you are using any new packages, install them in your virtual env and use this to update the requirements.txt ``` pip3 freeze > requirements.txt ```
+- Start the prefect orion UI server ``` screen -S prefect-server -dm prefect server start ```
+- To execute your flow you need to run an agent in different terminal ``` prefect agent start -q default ```. This will run the default queue worker.
+
+# prefect cli tools
+
+- Run the help command to understand the options ``` python3 main.py -h ```
+- Setup the organization config in ```config.json``` . The boilerplate for this is available in ``` configExample.json ```
+- Run the main file to deploy your organization ``` python3 main.py --deploy <org_name> ``` based on the config entered in ```config.json```
+- To reset the deployments after updates in config run ``` python3 main.py --deploy <org_name> --reset yes ```

configExample.json

@@ -0,0 +1,14 @@
+{
+    "stir": [
+        {
+            "connection_ids": ["<connection_ids>"],
+            "dbt_dir": null,
+            "schedule": "0 */2 * * *"
+        },
+        {
+            "connection_ids": ["<connection_ids>"],
+            "dbt_dir": "/Path/to/dbt/code/repo",
+            "schedule": "0 */2 * * *"
+        }
+    ]
+}

flows/Flow.py

@@ -0,0 +1,34 @@
+from prefect import flow
+from typing import List, Union
+from tasks.Airbyte import Airbyte
+from tasks.Dbt import Dbt
+from pydantic import BaseModel
+
+class Flow(BaseModel):
+    airbytes: Union[List[Airbyte], None]
+    dbt: Union[Dbt, None]
+
+@flow(flow_run_name='airbyte_flow')
+def airbyte_flow(flow: Flow):
+    for airbyte in flow.airbytes:
+        airbyte.sync()
+
+@flow(flow_run_name='dbt_flow')
+def dbt_flow(flow: Flow):
+    flow.dbt.pull_dbt_repo()
+    flow.dbt.dbt_deps()
+    flow.dbt.dbt_source_snapshot_freshness()
+    flow.dbt.dbt_run()
+    flow.dbt.dbt_test()
+
+@flow(flow_run_name='airbyte_dbt_flow')
+def airbyte_dbt_flow(flow: Flow):
+
+    for airbyte in flow.airbytes:
+        airbyte.sync()
+
+    flow.dbt.pull_dbt_repo()
+    flow.dbt.dbt_deps()
+    flow.dbt.dbt_source_snapshot_freshness()
+    flow.dbt.dbt_run()
+    flow.dbt.dbt_test()

main.py

@@ -1,62 +1,74 @@
-from prefect import flow, task
-from prefect_airbyte.connections import trigger_sync
-from prefect_dbt.cli.commands import trigger_dbt_cli_command
-from prefect_shell import shell_run_command
+import sys, getopt, argparse, json, os
 from dotenv import load_dotenv
-import os
+from pathlib import Path
+import asyncio
+from sqlalchemy import create_engine
+from prefect.settings import PREFECT_API_DATABASE_CONNECTION_URL, PREFECT_API_URL
+from sqlalchemy.ext.asyncio import create_async_engine, async_session, AsyncSession
+from sqlalchemy.orm import sessionmaker
 
+from organization.Organization import Organization
+
+# Load env
 load_dotenv()
 
-@flow(name="airbyte-sync")
-def run_airbyte_sync():
-      trigger_sync(
-            connection_id=os.getenv('DOST_AIRBYTE_CONNECTION'),
-            poll_interval_s=3,
-            status_updates=True
-      )
-      return 1
+# Load config
+async def main():
 
-@flow(name="github-pull")
-def pull_dost_github_repo():
-      shell_run_command('rm -rf dbt && git clone '+ os.getenv('DOST_GITHUB_URL'))
-      return 1
+    try:
+        config = None
+        if not Path('config.json').is_file():
+            raise Exception('Config file not found')
 
-@flow(name="dbt-transform")
-def run_dbt_transform():
-      trigger_dbt_cli_command(
-            command="dbt deps", project_dir='dbt'
-      )
-      trigger_dbt_cli_command(
-            command="dbt run", project_dir='dbt'
-      )
-      return 1
+        with open('config.json') as f:
+            config = json.load(f)
 
-@flow(name="orchestration-flow")
-def run_flow():
+        parser = argparse.ArgumentParser(description='Prefect deployment of NGOs')
+        parser.add_argument(
+            '--deploy',
+            required=True,
+            choices=['stir', 'sneha', 'shri'],
+            help='please enter the name of the NGO',
+            metavar='<org_name>'
+        )
+        parser.add_argument(
+            '--reset',
+            required=False,
+            choices=['yes', 'no'],
+            default='no',
+            help='resetting the deployments will remove all deployments and create fresh ones',
+            metavar='<yes or no>'
+        )
+        args = parser.parse_args()
 
-      #syncing airbyte
-      run_airbyte_sync()
+        if args.deploy not in config:
+            raise Exception(f'Config for {args.deploy} org not found')
+
+        # cli args
+        org_name = args.deploy
+        reset = args.reset
 
-      # pull dbt repo
-      pull_dost_github_repo()
+        # create a datbase session
+        url = PREFECT_API_DATABASE_CONNECTION_URL.value()
+        engine = create_async_engine(
+            url,
+            echo=True,
+        )
+        session = sessionmaker(engine, class_=AsyncSession)
 
-      #dbt transform
-      run_dbt_transform()
+        organization = Organization(name=org_name, session=session(), pipelines=config[org_name])
 
-@flow(name="orchestrate-airbyte")
-def run_airbyte_flow():
+        await organization.deploy(reset == 'yes')
+
+        await organization.close_session()
 
-      #syncing airbyte
-      run_airbyte_sync()
+    except Exception as e:
 
-@flow(name="orchestrate-dbt")
-def run_dbt_flow():
+        print(e)
 
-      # pull dbt repo
-      pull_dost_github_repo()
+if __name__ == "__main__":
 
-      #dbt transform
-      run_dbt_transform()
+    asyncio.run(main())
+
 
-if __name__ == "__main__":
-      run_dbt_flow()
+