Orchestrator course

workshops/orchestrator_course/dlt_pipeline.py

@@ -0,0 +1,80 @@
+import dlt
+from dlt.sources.rest_api import RESTAPIConfig, rest_api_source
+
+config: RESTAPIConfig = {
+    "client": {
+        "base_url": "https://api.github.com",
+        "auth": {
+            "token": dlt.secrets["sources.access_token"], 
+        },
+        "headers": {
+            "Accept": "application/vnd.github+json",
+            "X-GitHub-Api-Version": "2022-11-28"
+        },
+        "paginator": "header_link" 
+    },
+    "resources": [  
+        {
+            "name": "repos",
+            "endpoint": {
+                "path": "orgs/dlt-hub/repos"
+            },
+        },
+        {
+          "name": "contributors",
+          "endpoint": {
+            "path": "repos/dlt-hub/dlt/contributors",
+          },
+        },
+        {
+          "name": "issues",
+          "endpoint": {
+            "path": "repos/dlt-hub/dlt/issues",
+            "params": {
+                    "state": "open",  # Only get open issues
+                    "sort": "updated",
+                    "direction": "desc",
+                    "since": "{incremental.start_value}"  # For incremental loading
+                },
+                "incremental": {
+                    "cursor_path": "updated_at",
+                    "initial_value": pendulum.today().subtract(days=30).to_iso8601_string()
+                }
+          },
+        },
+        {
+          "name": "forks",
+          "endpoint": {
+            "path": "repos/dlt-hub/dlt/forks",
+            "params": {
+                    "sort": "oldest",      # Ensures ascending creation order
+                    "per_page": 100
+                },
+                "incremental": {           #backfill
+                    "cursor_path": "created_at",  
+                    "initial_value": "2025-07-01T00:00:00Z",
+                    "end_value": "2025-08-01T00:00:00Z",
+                    "row_order": "asc"
+                }
+          },
+        },
+        {
+          "name": "releases",
+          "endpoint": {
+            "path": "repos/dlt-hub/dlt/releases",
+          },
+        },
+    ],
+}
+
+github_source = rest_api_source(config)
+
+# pipeline = dlt.pipeline(
+#         pipeline_name="github_repos_issues",
+#         destination="duckdb",
+#         dataset_name="github_data",
+#         progress="log"  # Add logging as per rule recommendation
+#     )
+
+# load_info = pipeline.run(github_source)
+# print(load_info)

workshops/orchestrator_course/example_pipeline.py

@@ -0,0 +1,71 @@
+
+import dlt
+from dlt.sources.rest_api import RESTAPIConfig, rest_api_source
+
+config: RESTAPIConfig = {
+    "client": {
+        "base_url": "https://api.github.com",
+        "auth": {
+            "token": dlt.secrets["sources.access_token"], 
+        },
+        "headers": {
+            "Accept": "application/vnd.github+json",
+            "X-GitHub-Api-Version": "2022-11-28"
+        },
+        "paginator": "header_link" 
+    },
+    "resources": [  
+        {
+            "name": "repos",
+            "endpoint": {
+                "path": "orgs/dlt-hub/repos"
+            },
+        },
+        {
+          "name": "contributors",
+          "endpoint": {
+            "path": "repos/dlt-hub/dlt/contributors",
+          },
+        },
+        {
+          "name": "issues",
+          "endpoint": {
+            "path": "repos/dlt-hub/dlt/issues",
+            "params": {
+                    "state": "open",  # Only get open issues
+                    "sort": "updated",
+                    "direction": "desc",
+                    "since": "{incremental.start_value}"  # For incremental loading
+                },
+                "incremental": {
+                    "cursor_path": "updated_at",
+                    "initial_value": "2025-03-01T00:00:00Z",
+                }
+          },
+        },
+        {
+          "name": "forks",
+          "endpoint": {
+            "path": "repos/dlt-hub/dlt/forks",
+            "params": {
+                    "sort": "oldest",      # Ensures ascending creation order
+                    "per_page": 100
+                },
+                "incremental": {           #backfill
+                    "cursor_path": "created_at",  
+                    "initial_value": "2025-07-01T00:00:00Z",
+                    "end_value": "2025-08-01T00:00:00Z",
+                    "row_order": "asc"
+                }
+          },
+        },
+        {
+          "name": "releases",
+          "endpoint": {
+            "path": "repos/dlt-hub/dlt/releases",
+          },
+        },
+    ],
+}
+
+github_source = rest_api_source(config)

workshops/orchestrator_course/prefect_bckfil_flow.py

@@ -0,0 +1,93 @@
+from prefect import flow, task
+from prefect.blocks.system import Secret
+import os
+import dlt
+from prefect_github import GitHubCredentials
+from prefect.task_runners import ThreadPoolTaskRunner
+from datetime import datetime, timedelta, timezone
+from dlt.sources.rest_api import rest_api_source
+
+
+def _set_env(md_db: str):
+    """
+    Sets environment variables required by dlt pipelines.
+    - GitHub PAT (loaded from Prefect block "github-pat") for API auth
+    - MotherDuck token (loaded from Prefect block "motherduck-token") for destination auth
+    - Destination DB name (MotherDuck database)
+
+    Args:
+        md_db (str): Name of the MotherDuck database to write into.
+    """
+    # GitHub PAT (GitHubCredentials.token is SecretStr -> use .get_secret_value())
+    pat = GitHubCredentials.load("github-pat").token.get_secret_value()
+    os.environ["SOURCES__ACCESS_TOKEN"] = pat  # dlt.secrets["sources.access_token"]
+    # MotherDuck auth + DB name
+    md_token = Secret.load("motherduck-token").get()
+    os.environ["MOTHERDUCK_TOKEN"] = md_token
+    os.environ["DESTINATION__MOTHERDUCK__CREDENTIALS__PASSWORD"] = md_token
+    os.environ["DESTINATION__MOTHERDUCK__CREDENTIALS__DATABASE"] = md_db
+
+@task(retries=2, log_prints=True)
+def run_resource(resource_name: str, md_db: str, start_date: str = None, end_date: str = None):
+    """
+    Runs a specific resource from the GitHub API using the example pipeline configuration.
+    Supports backfilling (for forks resource) by setting start_date and end_date.
+
+    Args:
+        resource_name (str): Name of the GitHub resource to run.
+        md_db (str): Name of the MotherDuck database to write into.
+        start_date (str): ISO 8601 formatted start date for backfilling (for forks resource).
+        end_date (str): ISO 8601 formatted end date for backfilling (for forks resource).
+    """
+    # Ensure environment vars for dlt are set for this task execution
+    _set_env(md_db)
+    import example_pipeline,copy
+    # Copy pipeline config so modifications don’t leak between tasks
+    cfg = copy.deepcopy(example_pipeline.config)
+    # If the resource is "forks" and a backfill window is provided, inject it dynamically
+    if resource_name == "forks" and start_date and end_date:
+        for res in cfg["resources"]:
+            if res["name"] == "forks":
+                res["endpoint"]["incremental"]["initial_value"] = start_date
+                res["endpoint"]["incremental"]["end_value"] = end_date
+    # pick just one resource from your dlt source
+    src = rest_api_source(cfg).with_resources(resource_name)
+    # unique pipeline per resource avoids dlt state clashes
+    pipe = dlt.pipeline(
+        pipeline_name=f"rest_api_github__{resource_name}",
+        destination="motherduck",
+        dataset_name="rest_api_data_mlt_backfill",
+        progress="log",
+    )
+    # Run extraction + load into destination
+    info = pipe.run(src)
+    print(f"{resource_name} -> {info}")
+    return info
+
+@flow(task_runner=ThreadPoolTaskRunner(max_workers=5), log_prints=True)
+def main(md_db: str = "dlt_test"):
+    """
+    Main Prefect flow that runs all GitHub resources in parallel.
+
+    Args:
+        md_db (str, optional): MotherDuck database name. Defaults to "dlt_test".
+    """
+    # Get today's date in UTC and subtract one day to get yesterday's date
+    end_dt = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0)
+    start_dt = end_dt - timedelta(days=1)
+
+    start_iso = start_dt.isoformat()  # Outputs: 2025-08-30T00:00:00+00:00
+    end_iso = end_dt.isoformat() 
+
+    # Launch tasks concurrently (repos, contributors, issues, forks, releases)
+    a = run_resource.submit("repos", md_db)
+    b = run_resource.submit("contributors", md_db)
+    c = run_resource.submit("issues", md_db)
+    d = run_resource.submit("forks", md_db, start_date=start_iso, end_date=end_iso)
+    e = run_resource.submit("releases", md_db)
+
+    # Wait for all tasks to complete and return results
+    return a.result(), b.result(), c.result(), d.result(), e.result()
+
+if __name__ == "__main__":
+    main()