# Develop and ship a pipeline (/pipelines/development-workflow)

Work on a branch, try your DAGs as a development copy, and merge to put them in production.

Pipeline code lives in a git repository. A pull request gives your branch its own copy of the DAGs in
Airflow, so you can run it against the real platform without touching production; merging puts it in
production. This page walks through that loop.

<Callout title="Prerequisite">
  You need write access to your pipeline repository. To see and run your development copy in
  Airflow, you need `can_view_pipelines` and `can_run_pipelines` — see the
  [permissions reference](/access/permissions-reference).
</Callout>

## The repository

A pipeline repository has a `databaas.yml` at its root and your Airflow code in one directory:

```yaml title="databaas.yml"
bundle:
  name: my-team
targets:
  dev:
    mode: development
  prod:
    mode: production
artifacts:
  dags: airflow/
```

- `bundle.name` names this repository's set of DAGs. Your Databaas team tells you which name to use;
  a missing or malformed name fails the deployment.
- `targets` lists where the code can go: a `development` target for branches and one `production`
  target for `main`.
- `artifacts.dags` is the directory that gets deployed.

The repository's [deploy pipeline](#the-deploy-pipeline) reads this file.

A typical layout under `airflow/`:

<Files>
  <Folder name="airflow" defaultOpen>
    <Folder name="dags" defaultOpen>
      <File name="open_meteo_forecast.py" />
    </Folder>
    <Folder name="open_meteo_pipeline" defaultOpen>
      <File name="__init__.py" />
      <File name="source.py" />
    </Folder>
    <Folder name="tests" defaultOpen>
      <File name="test_dags.py" />
    </Folder>
    <File name=".airflowignore" />
    <File name="pyproject.toml" />
  </Folder>
  <File name="databaas.yml" />
</Files>

- Keep the DAG file in `dags/` to orchestration only; put the logic in a package directly under
  `airflow/`.
- Add every package you create to both `pyproject.toml` and `.airflowignore`. Leaving it out of
  `.airflowignore` makes Airflow import the package looking for DAGs.

## Develop on a branch

<Steps>
  <Step>
    ### Create a branch

    Branch in the repository itself, not in a fork: pull requests from forks are skipped and get no
    development copy. Do not commit to `main`.

    ```bash
    git switch -c add-forecast-pipeline
    ```
  </Step>
  <Step>
    ### Test locally

    Sync the environment and run the tests before every push. A DAG that fails to import never
    appears in Airflow, and the tests catch that on your laptop.

    ```bash
    cd airflow
    uv sync
    uv run pytest
    ```

    To run one DAG end to end, call `dag.test()`, for example with
    `uv run python dags/open_meteo_forecast.py` if the file ends with
    `if __name__ == "__main__": open_meteo_forecast().test()`.
  </Step>
  <Step>
    ### Open a pull request

    Push the branch and open a pull request. The [deploy pipeline](#the-deploy-pipeline) checks
    your DAGs and, if they pass, a development copy of every DAG in the repository appears in
    Airflow shortly after. Every later push to the branch replaces it; a file you delete on
    the branch disappears from the copy too.
  </Step>
  <Step>
    ### Run your development copy

    In **Pipelines**, filter the DAG list on your tag and unpause your DAG to run it. See
    [find your development DAGs](#find-your-development-dags).
  </Step>
  <Step>
    ### Merge

    Merging to `main` replaces the production DAGs with the merged code. Closing or merging the pull
    request removes the development copy.
  </Step>
</Steps>

## The deploy pipeline

Your repository's CI deploys the DAGs with the `databaas` command from `databaas-utils`. It runs three
commands, all driven by `databaas.yml`:

| Command | When | Does |
| ------- | ---- | ---- |
| `databaas dag validate -t <target>` | Every push to a pull request or `main` | Imports every DAG file with Airflow, exactly as the target will receive it. A failure stops the deploy. |
| `databaas dag deploy -t <target>` | After validation passes | Replaces the target's DAGs with the directory in `artifacts.dags`. Unchanged files are skipped; deleted files are removed. |
| `databaas dag destroy -t dev` | When the pull request closes | Removes the branch's development copy. |

On a pull request the target is `dev` and the development copy is named after the branch; on `main`
it is `prod`. The full command reference is in the
[databaas-utils documentation](https://wolkwork.github.io/databaas-utils/dags/).

### What validation catches

`databaas dag validate` fails the pipeline, before anything is deployed, on:

- a DAG file that does not import, including a duplicate DAG id or one longer than 250 characters;
- a repository with no DAGs at all;
- on a branch, a DAG id that does not start with `dev_<key>_`. Such a DAG would replace the
  production DAG of the same id; build every DAG id with `target.dag_id(...)`.

Each problem names the file in your repository, and GitHub shows it as an annotation on that file.
See [validating before a deploy](https://wolkwork.github.io/databaas-utils/dags/#validating-before-a-deploy)
in the databaas-utils documentation for how the check mirrors Airflow.

Validation imports your DAG files, so it needs Airflow and everything your DAG files import at the
top. Keep heavy imports inside task bodies, and validation needs little more than Airflow itself. To
run the same check before you push, from the environment your tests use:

```bash
uv run --with "databaas-utils[deploy]==0.6.0" databaas dag validate -t dev
```

### Set up the repository

Your Databaas team gives you a DAG deployment key, the storage endpoint and the bucket name for your
environment. In the repository settings on GitHub, add:

| Name | Kind | Value |
| ---- | ---- | ----- |
| `DATABAAS_DAGS_KEY_ID` | Secret | The deployment key's id |
| `DATABAAS_DAGS_KEY_SECRET` | Secret | The deployment key's secret |
| `DATABAAS_DAGS_ENDPOINT` | Variable | The storage endpoint |
| `DATABAAS_DAGS_BUCKET` | Variable | The bucket name |

Then add the workflow:

```yaml title=".github/workflows/deploy-dags.yaml"
name: Deploy DAGs

on:
  pull_request:
    types: [opened, synchronize, reopened, closed]
  push:
    branches: [main]

permissions:
  contents: read

env:
  DATABAAS_UTILS: databaas-utils[deploy]==0.6.0
  AIRFLOW_CONSTRAINTS: https://raw.githubusercontent.com/apache/airflow/constraints-3.1.8/constraints-3.12.txt
  AWS_ENDPOINT_URL: ${{ vars.DATABAAS_DAGS_ENDPOINT }}
  DATABAAS_DAGS_BUCKET: ${{ vars.DATABAAS_DAGS_BUCKET }}
  TARGET: ${{ github.event_name == 'push' && 'prod' || 'dev' }}

jobs:
  validate:
    if: github.event.action != 'closed'
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: astral-sh/setup-uv@v6
      # Add a --with for each package your DAG files import at the top.
      - run: |
          uvx --python 3.12 --from "$DATABAAS_UTILS" \
            --with apache-airflow==3.1.8 -c "$AIRFLOW_CONSTRAINTS" \
            databaas dag validate -t "$TARGET"

  deploy:
    needs: validate
    if: >
      github.event_name == 'push' ||
      github.event.pull_request.head.repo.full_name == github.repository
    runs-on: ubuntu-latest
    concurrency:
      group: deploy-dags-${{ github.head_ref || 'main' }}
      cancel-in-progress: false
    env:
      AWS_ACCESS_KEY_ID: ${{ secrets.DATABAAS_DAGS_KEY_ID }}
      AWS_SECRET_ACCESS_KEY: ${{ secrets.DATABAAS_DAGS_KEY_SECRET }}
    steps:
      - uses: actions/checkout@v4
      - uses: astral-sh/setup-uv@v6
      - run: uvx --from "$DATABAAS_UTILS" databaas dag deploy -t "$TARGET"

  cleanup:
    if: >
      github.event.action == 'closed' &&
      github.event.pull_request.head.repo.full_name == github.repository
    runs-on: ubuntu-latest
    concurrency:
      group: deploy-dags-${{ github.head_ref }}
      cancel-in-progress: false
    env:
      AWS_ACCESS_KEY_ID: ${{ secrets.DATABAAS_DAGS_KEY_ID }}
      AWS_SECRET_ACCESS_KEY: ${{ secrets.DATABAAS_DAGS_KEY_SECRET }}
    steps:
      - uses: actions/checkout@v4
      - uses: astral-sh/setup-uv@v6
      - run: uvx --from "$DATABAAS_UTILS" databaas dag destroy -t dev
```

- **Pin `databaas-utils` and Airflow.** Use the Airflow version Databaas runs, with Airflow's
  constraints file: without it, a newer SQLAlchemy breaks the Airflow import.
- **Pull requests from forks** are validated but not deployed: GitHub gives them no secrets.
- **Steps that build files into `artifacts.dags`**, such as `dbt parse` for a Cosmos manifest, go
  before the `databaas` command in both the `validate` and the `deploy` job.
- `uvx` installs `databaas-utils` on the CI runner only, as a tool. It is not one of your pipeline's
  requirements, so [the rule against installing it](/pipelines/authoring#what-databaas-utils-gives-you)
  still holds.

<Callout title="The deployment key">
  `databaas dag deploy` and `destroy` authenticate only with the key in `AWS_ACCESS_KEY_ID` and
  `AWS_SECRET_ACCESS_KEY`, never with your own login. Keep the key in repository secrets: on a laptop
  without it, the command stops and says which variable is missing. See
  [credentials](https://wolkwork.github.io/databaas-utils/dags/#credentials) in the databaas-utils
  documentation.
</Callout>

For Bitbucket Pipelines, see the
[Bitbucket example](https://wolkwork.github.io/databaas-utils/dags/#ci) in the databaas-utils
documentation.

## Find your development DAGs

Development DAGs run in the same Airflow as production and are told apart three ways:

- **Prefixed**: `dev_<key>_<name>` instead of `<name>`. The key is derived from your branch name.
- **Tagged**: `dev:<key>`. Filter the DAG list on this tag to see only yours.
- **Created paused.** Unpause yours when you want it to run.

Each development DAG's docs end with a **Deployed from** line: the repository, branch, commit and
deploy time. Check it when you are unsure whether you are looking at your latest push.

Production DAGs have no prefix and no `dev:` tag, and are not paused: they start running on their
schedule once merged.

The prefix, tag and pause state come from `target` in your DAG file; see
[deployment identity](/pipelines/authoring#deployment-identity-target).

## Keep a branch off production data

<Callout type="warn">
  A development DAG reads and writes the **real** catalog and storage. Nothing about the development
  copy makes a write a rehearsal.
</Callout>

Pass every name that identifies data through `scoped()`, so a branch writes to its own namespace and
tables instead of production's:

```python
from databaas_utils.airflow import scoped

NAMESPACE = scoped("open_meteo")  # "open_meteo" in production, "open_meteo_dev_<key>" on a branch
```

Creating that branch namespace needs a create grant on the warehouse for your pipeline's identity —
see [which identity a pipeline runs as](/pipelines/authoring#which-identity-a-pipeline-runs-as).

Data written under a scoped name is **not** removed when the pull request closes. Drop the
namespaces and tables your branch created once you no longer need them.

## Run tests without a deployment

A local checkout is not a deployment. Outside one, `scoped()` returns the production name, and
reading any `target` attribute raises a `RuntimeError`. For `dag.test()` and pytest, put a
production-shaped target file next to your DAGs and keep it out of git:

```bash
echo '{"mode": "production", "bundle": "local"}' > dags/_databaas_target.json
echo '_databaas_target.json' >> .gitignore
```

<Callout type="error">
  Never commit `_databaas_target.json`. Each deployment gets its own; a committed copy makes every
  branch deploy behave as production.
</Callout>

Without a `databaas_lakehouse` connection, every lakehouse helper raises `ConnectionNotConfigured`.
`get_store(local_root="/tmp/lake")` and `get_zarr_group("t", local_root="/tmp/lake")` write to a
local folder instead, as an explicit choice. `get_catalog` and `get_index` have no local mode: catch
`ConnectionNotConfigured` yourself if a laptop run should skip them.

## Mistakes worth knowing before your first pull request

| Mistake | How it shows up |
| ------- | --------------- |
| `from airflow.sdk import CronTriggerTimetable` | Import error in Airflow. Import timetables from `airflow.timetables`. |
| A package left out of `.airflowignore` | Airflow imports your package and tests looking for DAGs. |
| A lakehouse helper imported at the top of the DAG file | Every parse pays for it, and a broken import breaks the whole DAG. |
| A large payload returned from a task | Fine at a few hundred rows, fails at hundreds of megabytes. Store the file and pass its URI. |
| A task that is not idempotent | A retry silently doubles the rows. |
| A data name not passed through `scoped()` | Branch data shows up in a production table. |
| A DAG id not built with `target.dag_id(...)` | `databaas dag validate` fails the pull request. |