Schedule lineage-aware pipeline runs with agents, all in Python

dlthub init --name my_workspace
uv add "dlt[hub]"
dlthub login
dlthub workspace connect

"""Jaffle Shop -- ingest and transform customer data"""

from jaffle_ingestion import ingest_jaffle
from jaffle_transformations import transform_jaffle

__all__ = ["ingest_jaffle", "transform_jaffle"]

@run.pipeline("github_pipeline", trigger=trigger.every("5m"))
def load_commits():
    ...


@run.pipeline("jaffle_transform", trigger=ingest_jaffle.success)
def transform_jaffle():
    ...

dlthub deploy --dry-run
dlthub local run load_commits
dlthub deploy

dlthub job list "schedule:*"
dlthub job logs load_commits -f
dlthub job cancel "tag:ingest"

from dlt.hub.run import TJobRunContext

USGS_EPOCH = datetime(2025, 1, 1, tzinfo=timezone.utc)


@run.job(expose={"tags": ["backfill"]}, refresh="always")
def backfill_usgs():
    """Cascade a refresh signal downstream. Loads no data."""


@run.pipeline(
    usgs_ing_pipeline,
    interval={"start": USGS_EPOCH},                  # where the data begins
    trigger=["*/3 * * * *", backfill_usgs.success],
    require={"timezone": "Europe/Berlin"},
)
def usgs_daily(run_context: TJobRunContext):
    if run_context["refresh"]:
        usgs_ing_pipeline.refresh = "drop_sources"
    _load_ingest(
        run_context["interval_start"],   # supplied by the scheduler
        run_context["interval_end"],
        ["earthquakes", "feeds_summary"],
    )

import dlt
from airflow.decorators import dag
from dlt.common import pendulum
from dlt.helpers.airflow_helper import PipelineTasksGroup
from tenacity import Retrying, stop_after_attempt


@dag(schedule="@daily", start_date=pendulum.DateTime(2026, 1, 1), catchup=False)
def load_pipedrive():
    tasks = PipelineTasksGroup(
        pipeline_name="pipedrive",
        use_data_folder=False,     # True: stage working files in the Composer data bucket
        wipe_local_data=True,      # clean the pipeline working dir after the run
        use_task_logger=True,      # route dlt logs into the Airflow task log
        retry_policy=Retrying(stop=stop_after_attempt(3), reraise=True),
    )

    # import sources INSIDE the dag function — top-level code slows DAG parsing
    from pipedrive import pipedrive_source

    pipeline = dlt.pipeline(
        pipeline_name="pipedrive",
        destination="bigquery",
        dataset_name="pipedrive_data",
    )

    tasks.add_run(
        pipeline=pipeline,
        data=pipedrive_source(),
        decompose="serialize",
        trigger_rule="all_done",
        provide_context=True,
    )


load_pipedrive()

# fan every resource out in parallel
tasks.add_run(pipeline=pipeline, data=source, decompose="parallel", trigger_rule="all_done")

# or orchestrate at resource level: everything on the default cursor,
# but `activities` on its own start date
source = pipedrive_source()
source.resources["activities"].selected = False

activities_source = pipedrive_source(
    since_timestamp="2026-03-01 00:00:00Z"
).with_resources("activities")

tasks.add_run(pipeline=pipeline, data=source, decompose="serialize")
tasks.add_run(pipeline=pipeline, data=activities_source, decompose="serialize")

@dlt.resource(primary_key="id")
def tickets(
    zendesk_client,
    updated_at=dlt.sources.incremental[int](
        "updated_at",
        allow_external_schedulers=True,   # defer to the DAG's data interval
    ),
):
    for page in zendesk_client.get_pages(
        "/api/v2/incremental/tickets", "tickets",
        start_time=updated_at.start_value,
    ):
        yield page


@dag(
    schedule="@weekly",
    start_date=pendulum.datetime(2024, 1, 1, tz="UTC"),
    catchup=True,          # generate a run for every week since start_date
    max_active_runs=3,     # backfill three weeks at a time
)
def zendesk_backfill():
    ...

dlt deploy pipedrive_pipeline.py airflow-composer
# or, secrets as env vars instead of an Airflow Variable:
dlt deploy pipedrive_pipeline.py airflow-composer --secrets-format env