diff --git a/.dockerignore b/.dockerignore
new file mode 100644
index 0000000..a334663
--- /dev/null
+++ b/.dockerignore
@@ -0,0 +1,8 @@
+astro
+.git
+.env
+airflow_settings.yaml
+logs/
+.venv
+airflow.db
+airflow.cfg
diff --git a/.hyf/grader_lib.sh b/.hyf/grader_lib.sh
index 3142cfe..1ba13d3 100644
--- a/.hyf/grader_lib.sh
+++ b/.hyf/grader_lib.sh
@@ -7,11 +7,9 @@
 # and a set of common static-analysis checks derived from recurring
 # PR review patterns across cohort c55.
 #
-# blocker(): use for leaked-secret findings (a committed profiles.yml/.env,
-# a hardcoded password/connection string). It behaves like fail() for the
-# printed report, but also flips a flag that forces write_score() to report
-# pass=false regardless of the earned point total -- a leaked secret must
-# be fixed before the PR can pass, it cannot be "pointed around."
+# blocker(): use for findings that must fail the PR regardless of points
+# (leaked secrets, missing required evidence like screenshots). Behaves like
+# fail() in the printed report, but forces write_score() to pass=false.
 
 _grader_details=()
 _grader_blocker=false
@@ -38,7 +36,7 @@ write_score() {
   [[ "$score" -ge "$passing" ]] && pass_flag="true"
   if [[ "$_grader_blocker" == true ]]; then
     pass_flag="false"
-    echo "🚫 A blocker was found (leaked secret) -- forcing pass=false regardless of score." >&2
+    echo "🚫 A blocker was found -- forcing pass=false regardless of score." >&2
   fi
   cat > "$outfile" << JSON
 {
diff --git a/.hyf/test.sh b/.hyf/test.sh
index e693692..c596847 100755
--- a/.hyf/test.sh
+++ b/.hyf/test.sh
@@ -3,8 +3,9 @@
 # The DAG needs a running Astro/Airflow stack and a live Azure PostgreSQL
 # connection that CI cannot reach, so this checks file presence and code
 # patterns in dags/taxi_pipeline.py and the docs. The actual green run,
-# Screenshot files are presence-checked; content, backfill idempotency, and
-# shared-Airflow deploy are reviewed by a teacher.
+# Screenshot files are required (≥3): missing screenshots force pass=false.
+# Content of those shots, backfill idempotency, and shared-Airflow deploy
+# are still reviewed by a teacher.
 # Total points: 100. Passing score: 60.
 set -euo pipefail
 
@@ -218,9 +219,10 @@ shot_count=$(
 if [[ "$shot_count" -ge 3 ]]; then
   l6=$((l6 + 3)); pass "screenshots: found ${shot_count} image file(s) (need ≥3 for Graph + Grid/run + task log)"
 elif [[ "$shot_count" -gt 0 ]]; then
-  fail "screenshots: only ${shot_count} image file(s) — commit at least 3 (local Graph, green Grid/run, one task log; add shared-UI shot when the VM is up)"
+  # Screenshots are required evidence for teacher review — cannot pass without them.
+  blocker "screenshots: only ${shot_count} image file(s) — commit at least 3 (local Graph, green Grid/run, one task log; add shared-UI shot when the VM is up)"
 else
-  fail "screenshots: none found — commit Graph, Grid/run, and task-log images into the PR (any folder)"
+  blocker "screenshots: none found — commit Graph, Grid/run, and task-log images into the PR (any folder). Screenshots are required; a high code score without them still fails."
 fi
 score=$((score + l6))
 pass "Level 6: documentation + screenshots ($l6/10 pts)"
diff --git a/AI_ASSIST.md b/AI_ASSIST.md
index 171da98..88537c0 100644
--- a/AI_ASSIST.md
+++ b/AI_ASSIST.md
@@ -1,12 +1,18 @@
 # AI assistance log
 
 <!-- Document at least one point where you used an LLM on this assignment.
-     Never paste connection strings, passwords, or real data. Fill in each field. -->
+     Never paste connection strings, passwords, or real data. Replace TODO. -->
 
 ## Use 1
 
-**Prompt I sent:** _Replace this section._
+**Prompt I sent:**
 
-**What the model answered:** _Replace this section._
+Docker Desktop was stuck on “Starting the Docker Engine”, and Windows showed that the paging file was too small. I asked what safe steps I could try without deleting my project data.
 
-**What I kept, changed, or discarded, and why:** _Replace this section._
+**What the model answered:** 
+
+The model suggested restarting Windows, shutting down WSL, stopping unused Docker containers, and starting only one Astro project. It also warned me not to reset Docker to factory settings or remove Docker volumes because this could delete local Airflow data and settings.
+
+**What I kept, changed, or discarded, and why:** 
+
+I followed the safe steps: I restarted the computer, opened only the current project, and stopped containers from older projects. I did not uninstall Docker or delete any Docker volumes because I wanted to keep my local Airflow history and connection settings. I checked the result myself by running `docker ps` and reopening the Airflow interface. No passwords, connection strings, or private data were shared with the model.
diff --git a/RUNBOOK.md b/RUNBOOK.md
index 84bca1c..10b7875 100644
--- a/RUNBOOK.md
+++ b/RUNBOOK.md
@@ -1,22 +1,44 @@
 # RUNBOOK
 
-<!-- Fill in every section below. Another student should be able to
+<!-- Replace every TODO with real content. Another student should be able to
      operate your DAG from this file alone, without reading your Python. -->
 
 ## How to trigger the DAG manually
 
-_Replace this section._
+1. Start the local stack from the project root with `astro dev start`.
+2. Open the Airflow UI and confirm that the `azure_pg` connection exists.
+3. Open `taxi_pipeline`, unpause it, and click **Trigger**.
+4. In Trigger Options, pass a real past logical date such as `2024-01-01`.
+   If the UI does not expose a logical-date field, use this run configuration:
+   `{"target_date": "2024-01-01"}`.
+5. Open the new run and confirm the order `ingest_taxi_month → dbt_run → dbt_test`.
+
+Do not trigger the DAG for a future or unpublished TLC month. Such a run returns HTTP 403 because the parquet file does not exist yet.
 
 ## How to run a backfill
 
-_Replace this section._
+The DAG uses a monthly schedule, so seven assignment runs require seven months. In PowerShell run this as one line, without Bash backslashes:
+
+```powershell
+astro dev run backfill create --dag-id taxi_pipeline --from-date 2024-01-01 --to-date 2024-07-31 --max-active-runs 1
+```
+
+If the DAG is paused, create the backfill first and then unpause it in the UI. Wait until all runs finish before repeating the range. To repeat completed dates for the idempotency proof, use:
+
+```powershell
+astro dev run backfill create --dag-id taxi_pipeline --from-date 2024-01-01 --to-date 2024-07-31 --max-active-runs 1 --reprocess-behavior completed
+```
 
 ## How to inspect task logs
 
-_Replace this section._
+Open `taxi_pipeline → Runs → the failed run → the red task → Logs`. Read upward from the final generic message until the first concrete exception, such as `HTTPError`, `DatabaseError`, `Compilation Error`, or `Env var required but not provided`. Also check **Rendered Templates** for `dbt_run` and `dbt_test` to confirm the resolved database host, user, database, and schema. Passwords must remain redacted.
 
 ## Top 3 likely failures and first response
 
-1. _Replace: symptom, first check, fix_
-2. _Replace this section._
-3. _Replace this section._
+1. **`ingest_taxi_month` fails with HTTP 403/404.** Check the requested URL and logical date in the log. A future month or mistyped path is deterministic, so retries will not fix it. Trigger/backfill a published month and correct the URL or date range.
+2. **PostgreSQL connection or permission failure.** Confirm `AIRFLOW_STUDENT` in `.env`, then verify `azure_pg` in Admin → Connections. The login and schema name must match, for example role `halyna` writes to `airflow_halyna`. Do not commit credentials.
+3. **`dbt_run` fails.** Read the first dbt `Runtime Error` or `Compilation Error`, not only the final Bash exit code. Confirm `include/dbt_project/profiles.yml` exists, required `PG_*` variables are rendered, and `dbt deps` runs before `dbt run`.
+
+## Safe recovery and escalation
+
+Clear and retry one task only for a transient network or database interruption. Use a backfill after a code or business-logic fix that affects several partitions. If the shared scheduler, shared `azure_pg` connection, or shared VM is broken, do not edit shared settings; report the issue to the teacher.
\ No newline at end of file
diff --git a/dags/.airflowignore b/dags/.airflowignore
new file mode 100644
index 0000000..e69de29
diff --git a/dags/exampledag.py b/dags/exampledag.py
new file mode 100644
index 0000000..7c024cf
--- /dev/null
+++ b/dags/exampledag.py
@@ -0,0 +1,98 @@
+"""
+## Astronaut ETL example DAG
+
+This DAG queries the list of astronauts currently in space from the
+Open Notify API and prints each astronaut's name and flying craft.
+
+There are two tasks, one to get the data from the API and save the results,
+and another to print the results. Both tasks are written in Python using
+Airflow's TaskFlow API, which allows you to easily turn Python functions into
+Airflow tasks, and automatically infer dependencies and pass data.
+
+The second task uses dynamic task mapping to create a copy of the task for
+each Astronaut in the list retrieved from the API. This list will change
+depending on how many Astronauts are in space, and the DAG will adjust
+accordingly each time it runs.
+
+For more explanation and getting started instructions, see our Write your
+first DAG tutorial: https://www.astronomer.io/docs/learn/get-started-with-airflow
+
+![Picture of the ISS](https://www.esa.int/var/esa/storage/images/esa_multimedia/images/2010/02/space_station_over_earth/10293696-3-eng-GB/Space_Station_over_Earth_card_full.jpg)
+"""
+
+from airflow.sdk import Asset, dag, task
+from pendulum import datetime
+import requests
+
+
+# Define the basic parameters of the DAG, like schedule and start_date
+@dag(
+    start_date=datetime(2025, 4, 22),
+    schedule="@daily",
+    doc_md=__doc__,
+    default_args={"owner": "Astro", "retries": 3},
+    tags=["example"],
+)
+def example_astronauts():
+    # Define tasks
+    @task(
+        # Define an asset outlet for the task. This can be used to schedule downstream DAGs when this task has run.
+        outlets=[Asset("current_astronauts")]
+    )  # Define that this task updates the `current_astronauts` Asset
+    def get_astronauts(**context) -> list[dict]:
+        """
+        This task uses the requests library to retrieve a list of Astronauts
+        currently in space. The results are pushed to XCom with a specific key
+        so they can be used in a downstream pipeline. The task returns a list
+        of Astronauts to be used in the next task.
+        """
+        try:
+            r = requests.get("http://api.open-notify.org/astros.json")
+            r.raise_for_status()
+            number_of_people_in_space = r.json()["number"]
+            list_of_people_in_space = r.json()["people"]
+        except Exception:
+            print("API currently not available, using hardcoded data instead.")
+            number_of_people_in_space = 12
+            list_of_people_in_space = [
+                {"craft": "ISS", "name": "Oleg Kononenko"},
+                {"craft": "ISS", "name": "Nikolai Chub"},
+                {"craft": "ISS", "name": "Tracy Caldwell Dyson"},
+                {"craft": "ISS", "name": "Matthew Dominick"},
+                {"craft": "ISS", "name": "Michael Barratt"},
+                {"craft": "ISS", "name": "Jeanette Epps"},
+                {"craft": "ISS", "name": "Alexander Grebenkin"},
+                {"craft": "ISS", "name": "Butch Wilmore"},
+                {"craft": "ISS", "name": "Sunita Williams"},
+                {"craft": "Tiangong", "name": "Li Guangsu"},
+                {"craft": "Tiangong", "name": "Li Cong"},
+                {"craft": "Tiangong", "name": "Ye Guangfu"},
+            ]
+
+        context["ti"].xcom_push(
+            key="number_of_people_in_space", value=number_of_people_in_space
+        )
+        return list_of_people_in_space
+
+    @task
+    def print_astronaut_craft(greeting: str, person_in_space: dict) -> None:
+        """
+        This task creates a print statement with the name of an
+        Astronaut in space and the craft they are flying on from
+        the API request results of the previous task, along with a
+        greeting which is hard-coded in this example.
+        """
+        craft = person_in_space["craft"]
+        name = person_in_space["name"]
+
+        print(f"{name} is currently in space flying on the {craft}! {greeting}")
+
+    # Use dynamic task mapping to run the print_astronaut_craft task for each
+    # Astronaut in space
+    print_astronaut_craft.partial(greeting="Hello! :)").expand(
+        person_in_space=get_astronauts()  # Define dependencies using TaskFlow API syntax
+    )
+
+
+# Instantiate the DAG
+example_astronauts()
diff --git a/dags/taxi_pipeline.py b/dags/taxi_pipeline.py
index 1786d30..90cb822 100644
--- a/dags/taxi_pipeline.py
+++ b/dags/taxi_pipeline.py
@@ -10,55 +10,233 @@
 autograder fails while any NotImplementedError remains.
 """
 
+
+import io
 import os
-from datetime import datetime
+from datetime import datetime, timedelta
 from pathlib import Path
 
-from airflow.sdk import dag, task
+import pandas as pd
+import requests
+from sqlalchemy import text
+
+from airflow.providers.postgres.hooks.postgres import PostgresHook
+from airflow.providers.standard.operators.bash import BashOperator
+from airflow.sdk import dag, get_current_context, task
 
-# Your per-student schema. AIRFLOW_STUDENT is set in .env for local Astro dev;
-# on the shared VM it falls back to the dags/<name>/ directory name.
+
+# Local Astro reads AIRFLOW_STUDENT from .env.
+# Shared Airflow can use the name of the student's DAG folder.
 STUDENT = os.environ.get("AIRFLOW_STUDENT") or Path(__file__).parent.name
 SCHEMA = f"airflow_{STUDENT}"
+
 TLC_BASE = "https://d37ci6vzurychx.cloudfront.net/trip-data"
 
 
 def find_dbt_dir() -> str:
-    """Return the mounted dbt project path (Astro vs shared-VM install root)."""
+    """Return the mounted dbt project directory."""
+
     for candidate in (
-        "/usr/local/airflow/include/dbt_project",  # Astro CLI
-        "/opt/airflow/include/dbt_project",        # shared VM docker-compose
+        "/usr/local/airflow/include/dbt_project",
+        "/opt/airflow/include/dbt_project",
     ):
         if Path(candidate).is_dir():
             return candidate
+
     return "/usr/local/airflow/include/dbt_project"
 
 
 DBT_DIR = find_dbt_dir()
 
-
-@dag(
-    # Task 1 (see README): configure the decorator — schedule, start_date,
-    # catchup=False, max_active_runs=1, default_args retries, tags.
-    start_date=datetime(2024, 1, 1),
+DBT = (
+    "uvx --python 3.11 "
+    "--from 'dbt-core==1.10.*' "
+    "--with 'dbt-postgres==1.10.*' "
+    "dbt"
 )
-def taxi_pipeline():
-    @task
-    def ingest_taxi_month() -> int:
-        """Download one month of TLC green-taxi data and load it into
-        ``{SCHEMA}.raw_trips`` idempotently. Return the number of rows.
 
-        Task 2 and Task 3 (see README): derive the partition from the
-        logical date, DELETE-then-append that month, and filter the
-        parquet to the logical month before write (Gotcha #4).
-        """
-        raise NotImplementedError
+DBT_ENV = {
+    "PG_HOST": "{{ conn.azure_pg.host }}",
+    "PG_PORT": "{{ conn.azure_pg.port }}",
+    "PG_USER": "{{ conn.azure_pg.login }}",
+    "PG_PASSWORD": "{{ conn.azure_pg.password }}",
+    "PG_DBNAME": "{{ conn.azure_pg.schema }}",
+    "PG_SCHEMA": SCHEMA,
+}
+
+
+def parquet_url_for(ds: str) -> str:
+    """Return the TLC green-taxi parquet URL for a logical date."""
+
+    parsed_date = datetime.strptime(ds, "%Y-%m-%d")
+    year_month = parsed_date.strftime("%Y-%m")
 
-    # Task 2 (see README): add the two transform tasks, wire the full
-    # chain, and run the transform through the Chapter 4 command so it works
-    # on the image's Python. Task 4: add retry behaviour.
+    return f"{TLC_BASE}/green_tripdata_{year_month}.parquet"
 
-    ingest_taxi_month()
 
+def _partition_date_from_context() -> str:
+    """Return the partition date for the current Airflow run."""
 
-taxi_pipeline()
+    context = get_current_context()
+    dag_run = context["dag_run"]
+    run_id = context["run_id"]
+
+    # target_date використовується тільки для ручного запуску.
+    if run_id.startswith("manual__"):
+        target_date = (dag_run.conf or {}).get("target_date")
+
+        if not target_date:
+            raise ValueError(
+                'Manual run requires '
+                'conf={"target_date": "2024-01-01"}.'
+            )
+
+        datetime.strptime(target_date, "%Y-%m-%d")
+        return target_date
+
+    # Для scheduled і backfill кожен run використовує свій місяць.
+    data_interval_start = context.get("data_interval_start")
+
+    if data_interval_start is not None:
+        return data_interval_start.strftime("%Y-%m-%d")
+
+    if dag_run.logical_date is not None:
+        return dag_run.logical_date.strftime("%Y-%m-%d")
+
+    raise ValueError("Could not determine partition date.")
+
+@dag(
+    dag_id="taxi_pipeline",
+    schedule="@monthly",
+    start_date=datetime(2024, 1, 1),
+    catchup=False,
+    max_active_runs=1,
+    default_args={
+        "retries": 2,
+        "retry_delay": timedelta(minutes=2),
+    },
+    tags=["week12", "taxi", "orchestration"],
+)
+def taxi_pipeline():
+    """Download, transform, and test monthly taxi data."""
+
+    @task()
+    def ingest_taxi_month() -> int:
+        """Download and store one monthly partition idempotently."""
+
+        ds = _partition_date_from_context()
+        year_month = ds[:7]
+        url = parquet_url_for(ds)
+
+        print(f"Loading partition {year_month} from {url}")
+
+        # Download the monthly parquet file.
+        response = requests.get(url, timeout=60)
+        response.raise_for_status()
+
+        dataframe = pd.read_parquet(
+            io.BytesIO(response.content)
+        )
+
+        # Ensure that only the requested logical month is written.
+        dataframe["lpep_pickup_datetime"] = pd.to_datetime(
+            dataframe["lpep_pickup_datetime"]
+        )
+
+        dataframe = dataframe[
+            dataframe["lpep_pickup_datetime"]
+            .dt.strftime("%Y-%m")
+            .eq(year_month)
+        ].copy()
+
+        if dataframe.empty:
+            raise ValueError(
+                f"No rows found for partition {year_month}"
+            )
+
+        # Connect to Azure PostgreSQL.
+        hook = PostgresHook(
+            postgres_conn_id="azure_pg"
+        )
+        engine = hook.get_sqlalchemy_engine()
+
+        # Create the student's schema.
+        with engine.begin() as connection:
+            connection.execute(
+                text(
+                    f'CREATE SCHEMA IF NOT EXISTS "{SCHEMA}"'
+                )
+            )
+
+        # Create raw_trips if it does not exist yet.
+        dataframe.head(0).to_sql(
+            "raw_trips",
+            engine,
+            schema=SCHEMA,
+            if_exists="append",
+            index=False,
+        )
+
+        # Delete only the current month before reloading it.
+        # This makes repeated runs idempotent.
+        with engine.begin() as connection:
+            connection.execute(
+                text(
+                    f'DELETE FROM "{SCHEMA}".raw_trips '
+                    "WHERE to_char("
+                    "lpep_pickup_datetime, 'YYYY-MM'"
+                    ") = :year_month"
+                ),
+                {"year_month": year_month},
+            )
+
+        # Append the refreshed monthly partition.
+        dataframe.to_sql(
+            "raw_trips",
+            engine,
+            schema=SCHEMA,
+            if_exists="append",
+            index=False,
+            method="multi",
+            chunksize=1000,
+        )
+
+        row_count = len(dataframe)
+
+        print(
+            f"Loaded {row_count} rows into "
+            f"{SCHEMA}.raw_trips for {year_month}"
+        )
+
+        return row_count
+
+    dbt_run = BashOperator(
+        task_id="dbt_run",
+        bash_command=(
+            f"{DBT} deps "
+            f"--project-dir {DBT_DIR} "
+            f"--profiles-dir {DBT_DIR} && "
+            f"{DBT} run "
+            f"--project-dir {DBT_DIR} "
+            f"--profiles-dir {DBT_DIR}"
+        ),
+        env=DBT_ENV,
+        append_env=True,
+    )
+
+    dbt_test = BashOperator(
+        task_id="dbt_test",
+        bash_command=(
+            f"{DBT} test "
+            f"--project-dir {DBT_DIR} "
+            f"--profiles-dir {DBT_DIR}"
+        ),
+        env=DBT_ENV,
+        append_env=True,
+    )
+
+    ingest = ingest_taxi_month()
+    ingest >> dbt_run >> dbt_test
+
+
+taxi_pipeline()
\ No newline at end of file
diff --git a/evidence/backfill_7_runs.png b/evidence/backfill_7_runs.png
new file mode 100644
index 0000000..cdff7b7
Binary files /dev/null and b/evidence/backfill_7_runs.png differ
diff --git a/evidence/dbt_test_log.png b/evidence/dbt_test_log.png
new file mode 100644
index 0000000..5a41f3e
Binary files /dev/null and b/evidence/dbt_test_log.png differ
diff --git a/evidence/graph_view.png b/evidence/graph_view.png
new file mode 100644
index 0000000..ea82c5c
Binary files /dev/null and b/evidence/graph_view.png differ
diff --git a/evidence/local_green_run.png b/evidence/local_green_run.png
new file mode 100644
index 0000000..a5f8e31
Binary files /dev/null and b/evidence/local_green_run.png differ
diff --git a/include/dbt_project/.github/workflows/dbt-build.yml b/include/dbt_project/.github/workflows/dbt-build.yml
new file mode 100644
index 0000000..5ea8bcf
--- /dev/null
+++ b/include/dbt_project/.github/workflows/dbt-build.yml
@@ -0,0 +1,37 @@
+name: dbt build
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
+  schedule:
+    # Daily at 06:00 UTC — catches drift even when no PRs land.
+    - cron: "0 6 * * *"
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+
+      - name: Install dbt-core + dbt-postgres
+        run: pip install "dbt-core==1.11.*" "dbt-postgres==1.10.*"
+
+      - name: Render profiles.yml from template
+        run: |
+          mkdir -p ~/.dbt
+          sed "s/dev_<your_name>/dev_ci/" profiles.yml.example > ~/.dbt/profiles.yml
+
+      - name: dbt deps
+        run: |
+          if [ -f packages.yml ]; then dbt deps; fi
+
+      - name: dbt build
+        env:
+          PG_PASSWORD: ${{ secrets.PG_PASSWORD }}
+        run: dbt build --select +fct_trips --profiles-dir ~/.dbt
diff --git a/include/dbt_project/.gitignore b/include/dbt_project/.gitignore
new file mode 100644
index 0000000..e07de29
--- /dev/null
+++ b/include/dbt_project/.gitignore
@@ -0,0 +1,7 @@
+target/
+dbt_packages/
+logs/
+profiles.yml
+.env
+.DS_Store
+.user.yml
diff --git a/include/dbt_project/README.md b/include/dbt_project/README.md
index 67f2b5f..8389dbb 100644
--- a/include/dbt_project/README.md
+++ b/include/dbt_project/README.md
@@ -1,21 +1,100 @@
-# Put your dbt project here
+# nyc-taxi-dbt-reference
 
-Your `dbt_run` / `dbt_test` tasks run dbt against a project mounted at
-`include/dbt_project/`. Astro mounts this folder into the container at
-`/usr/local/airflow/include/dbt_project`.
+Reference dbt project for **HYF Data Track Week 10**. Mirrors the `nyc_taxi` project that the
+chapters in `Data Track/Week 10/` walk you through building.
 
-Use **one** of these:
+This repo is a **safety net**, not a starter. The chapters expect you to type each file
+yourself. If you fall behind or something breaks, check out the tag for the chapter you are
+on and diff your local copy against this one.
 
-1. **Your own Week 10 project.** Copy your `nyc-taxi-dbt` models, `dbt_project.yml`,
-   and `profiles.yml` into this directory.
-2. **The class reference project** (if your Week 10 project is not runnable):
+## Tags (one per chapter checkpoint)
 
-   ```bash
-   git clone https://github.com/lassebenni/nyc-taxi-airflow-reference /tmp/class-ref
-   cp -r /tmp/class-ref/include/dbt_project/. include/dbt_project/
-   ```
+| Tag                     | Chapter                                  | What you should have at this point                                  |
+| ----------------------- | ---------------------------------------- | ------------------------------------------------------------------- |
+| `ch2-dbt-setup-azure`   | Ch 2 — dbt Setup for Azure PostgreSQL    | `dbt_project.yml`, `profiles.yml.example`, `_sources.yml`, plain `stg_trips.sql` |
+| `ch3-sql-jinja`         | Ch 3 — SQL and Jinja Templating          | `stg_trips.sql` extended with `payment_type_label`, `tip_pct`, `fare_per_mile` |
+| `ch4-materializations`  | Ch 4 — Materializations & Layers         | `stg_zones.sql`, `fct_trips.sql`, materialization config in `dbt_project.yml` |
+| `ch5-dbt-tests`         | Ch 5 — dbt Tests                         | All schema YAML, singular test, unit test, `packages.yml`           |
+| `ch6-docs-extras`       | Ch 6 — Docs & Extras                     | Fleshed-out `_fct_trips.yml` descriptions, doc block, `mutable_zones` seed + snapshot |
 
-Document which one you used in `ASSIGNMENT_REPORT.md`.
+```bash
+git checkout ch4-materializations   # rewind to end of Chapter 4
+git checkout main                   # latest (= end of Chapter 6)
+```
 
-> The dbt tasks call dbt through `uvx --python 3.11`, so you do **not** put
-> `dbt-core` in `requirements.txt`. See Chapter 4 for the exact command.
+## Setup
+
+Prereqs: dbt-core 1.11 with `dbt-postgres` (see [Ch 2](https://github.com/hackyourfuture/datatrack/blob/main/Data%20Track/Week%2010/week_10__2_dbt_setup_azure.md) for install).
+
+```bash
+cp profiles.yml.example profiles.yml      # then edit `schema:` to dev_<your_name>
+export PG_PASSWORD='your-week-6-password'  # macOS/Linux/WSL
+just deps                                  # dbt deps  (only needed from v4 onward)
+just build                                 # dbt build --select +fct_trips
+```
+
+`profiles.yml` is gitignored — it stays local so the file never lands in version control with a real password.
+
+## Running from Airflow (Week 11)
+
+[Week 11 Chapter 4](https://github.com/hackyourfuture/datatrack/blob/main/Data%20Track/Week%2011/week_11__4_sequential_pipelines.md) runs this dbt project end-to-end from an Airflow `BashOperator`. Three concrete setup steps on top of the local-dev setup above:
+
+### 1. Copy the project into your Astro project's `include/` folder
+
+Astro mounts `include/` into every container. Putting the dbt project under `include/dbt_project/` is the convention Ch4's `DBT_DIR` constant expects:
+
+```bash
+# from your Astro project directory
+cp -r ../nyc-taxi-dbt-reference include/dbt_project
+
+# or, if you cloned this repo as a submodule:
+git submodule add git@github.com:hackyourfuture/nyc-taxi-dbt-reference include/dbt_project
+cd include/dbt_project && git checkout week-11-airflow && cd ../..
+```
+
+### 2. Pass the connection credentials through `BashOperator.env`
+
+The `profiles.yml.example` on this branch reads **four** env vars with sensible defaults for the shared class DB: `PG_HOST`, `PG_USER`, `PG_DBNAME`, `PG_SCHEMA`, plus `PG_PASSWORD` which has no default (security). Pull them from the `azure_pg` Airflow Connection you created in Ch4 and forward them to the bash command:
+
+```python
+DBT_DIR = "/usr/local/airflow/include/dbt_project"
+DBT_ENV = {
+    "PG_HOST": "{{ conn.azure_pg.host }}",
+    "PG_USER": "{{ conn.azure_pg.login }}",
+    "PG_PASSWORD": "{{ conn.azure_pg.password }}",
+    "PG_DBNAME": "{{ conn.azure_pg.schema }}",
+    "PG_SCHEMA": "airflow_taxi",   # team-shared schema, not per-student
+}
+
+dbt_run = BashOperator(
+    task_id="dbt_run",
+    bash_command=f"dbt run --project-dir {DBT_DIR} --profiles-dir {DBT_DIR}",
+    env=DBT_ENV,
+    append_env=True,   # inherit PATH etc. from the worker
+)
+dbt_test = BashOperator(
+    task_id="dbt_test",
+    bash_command=f"dbt test --project-dir {DBT_DIR} --profiles-dir {DBT_DIR}",
+    env=DBT_ENV,
+    append_env=True,
+)
+```
+
+No passwords in `profiles.yml`, no passwords in DAG code, all credentials flow through the Airflow Connection the way the rest of Ch4 teaches.
+
+### 3. Use a shared schema, not `dev_<your_name>`
+
+Airflow writes into a single schema that downstream dashboards will read from. Pick one team-shared name (the example above uses `airflow_taxi`) and match it to whatever schema the `load_raw_trips` task writes to. A per-student `dev_<your_name>` schema defeats the point of orchestrating a shared pipeline.
+
+### Differences from `main`
+
+Only two deliberate differences from the `main` branch:
+
+- `profiles.yml.example` reads `PG_HOST`, `PG_USER`, `PG_DBNAME`, `PG_SCHEMA` as env vars in addition to `PG_PASSWORD`, so the same file works for local dbt CLI (env vars fall back to the class defaults) and for Airflow (env vars come from the `azure_pg` connection).
+- This README gained the Airflow section above.
+
+All model code, tests, seeds, and snapshots are byte-identical to `main`. Students who finished Week 10 on `main` can switch to `week-11-airflow` without needing to redo any of the SQL work.
+
+## CI
+
+Every push runs `dbt build --select +fct_trips` against the shared Azure PostgreSQL instance. If the curriculum prose drifts away from runnable reality (test result counts change, columns get renamed, etc.) CI fails and the chapter gets fixed. See `.github/workflows/dbt-build.yml`.
diff --git a/include/dbt_project/dbt_project.yml b/include/dbt_project/dbt_project.yml
new file mode 100644
index 0000000..1da9e46
--- /dev/null
+++ b/include/dbt_project/dbt_project.yml
@@ -0,0 +1,21 @@
+name: 'nyc_taxi'
+version: '1.0.0'
+profile: 'nyc_taxi'
+
+model-paths: ["models"]
+analysis-paths: ["analyses"]
+test-paths: ["tests"]
+seed-paths: ["seeds"]
+macro-paths: ["macros"]
+snapshot-paths: ["snapshots"]
+
+clean-targets:
+  - "target"
+  - "dbt_packages"
+
+models:
+  nyc_taxi:
+    staging:
+      +materialized: view
+    marts:
+      +materialized: table
diff --git a/include/dbt_project/justfile b/include/dbt_project/justfile
new file mode 100644
index 0000000..8edd683
--- /dev/null
+++ b/include/dbt_project/justfile
@@ -0,0 +1,30 @@
+DBT := "dbt"
+
+# Show available recipes
+default:
+    @just --list
+
+# Verify dbt can connect to Azure PostgreSQL
+debug:
+    {{DBT}} debug --profiles-dir .
+
+# Install dbt packages from packages.yml (only needed from v4 onward)
+deps:
+    {{DBT}} deps
+
+# Build all models from sources through fct_trips, with tests interleaved
+build:
+    {{DBT}} build --select +fct_trips --profiles-dir .
+
+# Build models only (no tests) — useful while iterating locally
+run:
+    {{DBT}} run --select +fct_trips --profiles-dir .
+
+# Run every test in the project
+test:
+    {{DBT}} test --profiles-dir .
+
+# Drop and rebuild everything from a clean target/
+clean-build:
+    {{DBT}} clean
+    just build
diff --git a/include/dbt_project/models/marts/_fct_daily_borough_stats.yml b/include/dbt_project/models/marts/_fct_daily_borough_stats.yml
new file mode 100644
index 0000000..e1e0614
--- /dev/null
+++ b/include/dbt_project/models/marts/_fct_daily_borough_stats.yml
@@ -0,0 +1,56 @@
+version: 2
+
+models:
+  - name: fct_daily_borough_stats
+    description: |
+      Daily borough-level mart for NYC green taxi pickup activity.
+
+      Grain: one row per pickup_borough and pickup_date.
+
+      Source lineage: built from stg_trips joined with stg_zones.
+      stg_trips reads nyc_taxi.raw_trips.
+      stg_zones reads nyc_taxi.raw_zones.
+
+      Known caveats: stg_trips drops rows with missing pickup_location_id
+      or negative fare_amount. The mart uses an inner join to drop trips
+      whose pickup_location_id does not match a known zone. Real TLC labels
+      such as EWR, Unknown, and dirty NaN borough values remain in the mart.
+      The avg_tip_pct singular test uses WARN severity because very small
+      borough/day groups can legitimately exceed 1 due to outliers.
+
+    tests:
+      - dbt_utils.unique_combination_of_columns:
+          combination_of_columns:
+            - pickup_borough
+            - pickup_date
+
+    columns:
+      - name: pickup_borough
+        description: Borough label from stg_zones.borough for the trip pickup zone.
+        tests:
+          - not_null
+
+      - name: pickup_date
+        description: Calendar date derived from pickup_datetime. Unit is one local pickup day.
+        tests:
+          - not_null
+
+      - name: trip_count
+        description: Number of trips picked up in this borough on this date. Unit is trips.
+        tests:
+          - not_null
+
+      - name: total_fare
+        description: Sum of fare_amount for this borough and date. Unit is US dollars.
+        tests:
+          - not_null
+
+      - name: avg_tip_pct
+        description: Average tip percentage for this borough and date. Calculated as tip_amount divided by fare_amount.
+        tests:
+          - not_null
+
+      - name: avg_trip_distance
+        description: Average trip distance for this borough and date. Unit is miles.
+        tests:
+          - not_null
\ No newline at end of file
diff --git a/include/dbt_project/models/marts/_fct_trips.yml b/include/dbt_project/models/marts/_fct_trips.yml
new file mode 100644
index 0000000..4bccd80
--- /dev/null
+++ b/include/dbt_project/models/marts/_fct_trips.yml
@@ -0,0 +1,50 @@
+version: 2
+
+models:
+  - name: fct_trips
+    description: |
+      One row per completed NYC green taxi trip in January 2024, with
+      pickup/dropoff zone attributes folded in (OBT-style mart). Queried
+      directly by dashboards and ad-hoc analysis.
+
+      **Grain:** one row per trip.
+      **Source:** `public.raw_trips` joined to `public.raw_zones` on
+      `pickup_location_id` and `dropoff_location_id`.
+      **Not included:** trips where `pickup_location_id` is NULL (dropped
+      in `stg_trips`); duplicate rows from the TLC source are kept as-is
+      and surfaced by `dbt_utils.unique_combination_of_columns`.
+    columns:
+      - name: pickup_datetime
+        description: Wall-clock time the trip began (America/New_York, no timezone attached).
+        tests: [not_null]
+      - name: dropoff_datetime
+        description: Wall-clock time the trip ended.
+      - name: fare_amount
+        description: Metered fare in USD, not including tip, tolls, or surcharges.
+      - name: tip_amount
+        description: Tip in USD. Non-zero only when payment_type is credit card (1).
+      - name: trip_distance
+        description: Distance in miles as reported by the taximeter.
+      - name: tip_pct
+        description: |
+          `tip_amount / fare_amount`, rounded to 4 decimals. NULL when
+          `fare_amount` is 0 (voided trips, no-charge rides).
+      - name: fare_per_mile
+        description: |
+          `fare_amount / trip_distance`, rounded to 4 decimals. NULL when
+          `trip_distance` is 0 (data-quality anomalies).
+      - name: payment_type_label
+        description: |
+          Human-readable payment method from the TLC code. See the jinja
+          dictionary in `stg_trips.sql` for the 1-6 → label mapping.
+      - name: pickup_borough
+        description: |
+          NYC borough of the pickup zone, joined from `stg_zones.borough`.
+          Values: Manhattan, Brooklyn, Queens, Bronx, Staten Island, EWR,
+          Unknown, NaN, or NULL when `pickup_location_id` did not resolve.
+      - name: pickup_zone
+        description: Human-readable pickup-zone name from `stg_zones.zone`.
+      - name: dropoff_borough
+        description: NYC borough of the dropoff zone.
+      - name: dropoff_zone
+        description: Human-readable dropoff-zone name.
diff --git a/include/dbt_project/models/marts/fct_daily_borough_stats.sql b/include/dbt_project/models/marts/fct_daily_borough_stats.sql
new file mode 100644
index 0000000..774c650
--- /dev/null
+++ b/include/dbt_project/models/marts/fct_daily_borough_stats.sql
@@ -0,0 +1,28 @@
+-- Mart: daily borough trip statistics.
+-- Grain: one row per (pickup_borough, pickup_date).
+-- Used to answer: trip volume, revenue, tipping behaviour, and distance profile
+-- per borough per day for January 2024.
+
+WITH trips AS (
+    SELECT *
+    FROM {{ ref('stg_trips') }}
+),
+
+zones AS (
+    SELECT *
+    FROM {{ ref('stg_zones') }}
+)
+
+SELECT
+    z.borough as pickup_borough,
+    t.pickup_datetime::date as pickup_date,
+    count(*)::bigint as trip_count,
+    sum(t.fare_amount)::numeric as total_fare,
+    avg(t.tip_pct)::numeric as avg_tip_pct,
+    avg(t.trip_distance)::numeric as avg_trip_distance
+FROM trips t
+inner join zones z
+    on t.pickup_location_id = z.location_id
+group by
+    z.borough,
+    t.pickup_datetime::date
\ No newline at end of file
diff --git a/include/dbt_project/models/marts/fct_trips.sql b/include/dbt_project/models/marts/fct_trips.sql
new file mode 100644
index 0000000..adaaec1
--- /dev/null
+++ b/include/dbt_project/models/marts/fct_trips.sql
@@ -0,0 +1,20 @@
+{{ config(materialized='table') }}
+
+select
+    t.pickup_datetime,
+    t.dropoff_datetime,
+    t.fare_amount,
+    t.tip_amount,
+    t.trip_distance,
+    t.tip_pct,
+    t.fare_per_mile,
+    t.payment_type_label,
+    pz.borough as pickup_borough,
+    pz.zone as pickup_zone,
+    dz.borough as dropoff_borough,
+    dz.zone as dropoff_zone
+from {{ ref('stg_trips') }} t
+left join {{ ref('stg_zones') }} pz
+    on t.pickup_location_id = pz.location_id
+left join {{ ref('stg_zones') }} dz
+    on t.dropoff_location_id = dz.location_id
diff --git a/include/dbt_project/models/marts/fct_trips_docs.md b/include/dbt_project/models/marts/fct_trips_docs.md
new file mode 100644
index 0000000..2c1db8d
--- /dev/null
+++ b/include/dbt_project/models/marts/fct_trips_docs.md
@@ -0,0 +1,10 @@
+{% docs trip_grain %}
+
+One row per completed taxi trip. "Completed" means the TLC submitted the
+trip record to the public dataset; cancellations and trips in progress
+are not included. Duplicates exist in the source data (roughly 4 rows in
+January 2024 where every column is identical) and are kept as-is; see
+the `dbt_utils.unique_combination_of_columns` test results for the
+current count.
+
+{% enddocs %}
diff --git a/include/dbt_project/models/staging/_sources.yml b/include/dbt_project/models/staging/_sources.yml
new file mode 100644
index 0000000..6a55f80
--- /dev/null
+++ b/include/dbt_project/models/staging/_sources.yml
@@ -0,0 +1,12 @@
+version: 2
+
+sources:
+  - name: nyc_taxi
+    description: Raw NYC green taxi trip records and zone lookup, loaded in Week 9.
+    database: team1
+    schema: nyc_taxi
+    tables:
+      - name: raw_trips
+        description: One row per green taxi trip for January 2024 (~57K rows).
+      - name: raw_zones
+        description: NYC taxi zone lookup (265 rows mapping location IDs to boroughs).
diff --git a/include/dbt_project/models/staging/_stg_trips.yml b/include/dbt_project/models/staging/_stg_trips.yml
new file mode 100644
index 0000000..656a4fc
--- /dev/null
+++ b/include/dbt_project/models/staging/_stg_trips.yml
@@ -0,0 +1,52 @@
+version: 2
+
+models:
+  - name: stg_trips
+    description: Cleaned green taxi trips, one row per trip.
+    tests:
+      # Chapter 5 teaches this test at the default `error` severity to demonstrate
+      # how `dbt build` skips downstream models on a test failure. The January 2024
+      # raw_trips data contains 4 genuine duplicate rows (a TLC source-data issue),
+      # so the test always fails. In this reference repo we soften it to `warn` to
+      # keep CI green while still surfacing the count. When students follow the
+      # chapter on their own machine, they should leave it at the default.
+      - dbt_utils.unique_combination_of_columns:
+          combination_of_columns: [pickup_datetime, dropoff_datetime, pickup_location_id, fare_amount]
+          config:
+            severity: warn
+    columns:
+      - name: pickup_datetime
+        description: When the trip started.
+        tests:
+          - not_null
+      - name: pickup_location_id
+        description: TLC zone ID where the trip started.
+        tests:
+          - not_null
+          - relationships:
+              to: ref('stg_zones')
+              field: location_id
+              config:
+                severity: warn
+      - name: payment_type
+        description: TLC payment code (1-6).
+        tests:
+          - not_null:
+              severity: warn
+          - accepted_values:
+              values: [1, 2, 3, 4, 5, 6]
+
+unit_tests:
+  - name: payment_type_label_maps_known_codes
+    model: stg_trips
+    given:
+      - input: source('nyc_taxi', 'raw_trips')
+        rows:
+          - {payment_type: 1, pickup_datetime: '2024-01-01 08:00:00', pickup_location_id: 100, fare_amount: 10.0, tip_amount: 2.0, trip_distance: 2.0}
+          - {payment_type: 2, pickup_datetime: '2024-01-01 09:00:00', pickup_location_id: 100, fare_amount: 10.0, tip_amount: 0.0, trip_distance: 2.0}
+          - {payment_type: 6, pickup_datetime: '2024-01-01 10:00:00', pickup_location_id: 100, fare_amount: 10.0, tip_amount: 0.0, trip_distance: 2.0}
+    expect:
+      rows:
+        - {payment_type: 1, payment_type_label: 'Credit card'}
+        - {payment_type: 2, payment_type_label: 'Cash'}
+        - {payment_type: 6, payment_type_label: 'Voided trip'}
diff --git a/include/dbt_project/models/staging/_stg_zones.yml b/include/dbt_project/models/staging/_stg_zones.yml
new file mode 100644
index 0000000..14f8227
--- /dev/null
+++ b/include/dbt_project/models/staging/_stg_zones.yml
@@ -0,0 +1,13 @@
+version: 2
+
+models:
+  - name: stg_zones
+    description: One row per TLC taxi zone (265 zones total).
+    columns:
+      - name: location_id
+        description: TLC zone ID.
+        tests:
+          - unique
+          - not_null
+      - name: borough
+        description: NYC borough (Manhattan, Brooklyn, Queens, Bronx, Staten Island, EWR, Unknown).
diff --git a/include/dbt_project/models/staging/stg_trips.sql b/include/dbt_project/models/staging/stg_trips.sql
new file mode 100644
index 0000000..08cc069
--- /dev/null
+++ b/include/dbt_project/models/staging/stg_trips.sql
@@ -0,0 +1,36 @@
+{{ config(materialized='view') }}
+
+{% set payment_types = {
+    1: 'Credit card',
+    2: 'Cash',
+    3: 'No charge',
+    4: 'Dispute',
+    5: 'Unknown',
+    6: 'Voided trip'
+} %}
+
+select
+    pickup_datetime,
+    dropoff_datetime,
+    pickup_location_id,
+    dropoff_location_id,
+    fare_amount,
+    tip_amount,
+    trip_distance,
+    payment_type,
+    case
+        when fare_amount > 0 then round((tip_amount / fare_amount)::numeric, 4)
+        else null
+    end as tip_pct,
+    case
+        when trip_distance > 0 then round((fare_amount / trip_distance)::numeric, 4)
+        else null
+    end as fare_per_mile,
+    case payment_type
+        {% for code, label in payment_types.items() %}
+        when {{ code }} then '{{ label }}'
+        {% endfor %}
+        else 'Other'
+    end as payment_type_label
+from {{ source('nyc_taxi', 'raw_trips') }}
+where pickup_location_id is not null
diff --git a/include/dbt_project/models/staging/stg_zones.sql b/include/dbt_project/models/staging/stg_zones.sql
new file mode 100644
index 0000000..1e98897
--- /dev/null
+++ b/include/dbt_project/models/staging/stg_zones.sql
@@ -0,0 +1,6 @@
+select
+    location_id,
+    borough,
+    zone,
+    service_zone
+from {{ source('nyc_taxi', 'raw_zones') }}
diff --git a/include/dbt_project/package-lock.yml b/include/dbt_project/package-lock.yml
new file mode 100644
index 0000000..abff2f9
--- /dev/null
+++ b/include/dbt_project/package-lock.yml
@@ -0,0 +1,5 @@
+packages:
+  - name: dbt_utils
+    package: dbt-labs/dbt_utils
+    version: 1.3.3
+sha1_hash: e6424ba9e5a22487e47f023803aa4f0411946808
diff --git a/include/dbt_project/packages.yml b/include/dbt_project/packages.yml
new file mode 100644
index 0000000..b26b1ff
--- /dev/null
+++ b/include/dbt_project/packages.yml
@@ -0,0 +1,3 @@
+packages:
+  - package: dbt-labs/dbt_utils
+    version: [">=1.1.0", "<2.0.0"]
diff --git a/include/dbt_project/profiles.yml.example b/include/dbt_project/profiles.yml.example
new file mode 100644
index 0000000..ca59c8e
--- /dev/null
+++ b/include/dbt_project/profiles.yml.example
@@ -0,0 +1,13 @@
+nyc_taxi:
+  target: dev
+  outputs:
+    dev:
+      type: postgres
+      host: "{{ env_var('PG_HOST', 'hyf-data-pg.postgres.database.azure.com') }}"
+      port: 5432
+      user: "{{ env_var('PG_USER', 'hyfadmin') }}"
+      password: "{{ env_var('PG_PASSWORD') }}"
+      dbname: "{{ env_var('PG_DBNAME', 'team1') }}"
+      schema: "{{ env_var('PG_SCHEMA', 'dev_<your_name>') }}"   # ⚠️ Local dev: replace <your_name> with your first name, lowercase. Airflow: pass PG_SCHEMA via BashOperator env (see README § Running from Airflow).
+      sslmode: require
+      threads: 4
diff --git a/include/dbt_project/seeds/mutable_zones.csv b/include/dbt_project/seeds/mutable_zones.csv
new file mode 100644
index 0000000..0a1a11a
--- /dev/null
+++ b/include/dbt_project/seeds/mutable_zones.csv
@@ -0,0 +1,11 @@
+location_id,borough,zone,service_zone
+1,EWR,Newark Airport,EWR
+4,Manhattan,Alphabet City,Yellow Zone
+13,Manhattan,Battery Park City,Yellow Zone
+87,Manhattan,Financial District North,Yellow Zone
+132,Queens,JFK Airport,Airports
+138,Queens,LaGuardia Airport,Airports
+161,Manhattan,Midtown Center,Yellow Zone
+236,Manhattan,Upper East Side North,Yellow Zone
+264,Unknown,NV,N/A
+265,Unknown,Outside of NYC,N/A
diff --git a/include/dbt_project/snapshots/mutable_zones_snapshot.sql b/include/dbt_project/snapshots/mutable_zones_snapshot.sql
new file mode 100644
index 0000000..f01022e
--- /dev/null
+++ b/include/dbt_project/snapshots/mutable_zones_snapshot.sql
@@ -0,0 +1,12 @@
+{% snapshot mutable_zones_snapshot %}
+
+{{ config(
+    target_schema=target.schema ~ '_snapshots',
+    unique_key='location_id',
+    strategy='check',
+    check_cols=['borough', 'zone', 'service_zone']
+) }}
+
+select * from {{ ref('mutable_zones') }}
+
+{% endsnapshot %}
diff --git a/include/dbt_project/tests/assert_pickup_before_dropoff.sql b/include/dbt_project/tests/assert_pickup_before_dropoff.sql
new file mode 100644
index 0000000..5092eba
--- /dev/null
+++ b/include/dbt_project/tests/assert_pickup_before_dropoff.sql
@@ -0,0 +1,6 @@
+select
+    pickup_datetime,
+    dropoff_datetime,
+    pickup_location_id
+from {{ ref('stg_trips') }}
+where pickup_datetime > dropoff_datetime
diff --git a/tests/dags/test_dag_example.py b/tests/dags/test_dag_example.py
new file mode 100644
index 0000000..6ff3552
--- /dev/null
+++ b/tests/dags/test_dag_example.py
@@ -0,0 +1,83 @@
+"""Example DAGs test. This test ensures that all Dags have tags, retries set to two, and no import errors. This is an example pytest and may not be fit the context of your DAGs. Feel free to add and remove tests."""
+
+import os
+import logging
+from contextlib import contextmanager
+import pytest
+from airflow.models import DagBag
+
+
+@contextmanager
+def suppress_logging(namespace):
+    logger = logging.getLogger(namespace)
+    old_value = logger.disabled
+    logger.disabled = True
+    try:
+        yield
+    finally:
+        logger.disabled = old_value
+
+
+def get_import_errors():
+    """
+    Generate a tuple for import errors in the dag bag
+    """
+    with suppress_logging("airflow"):
+        dag_bag = DagBag(include_examples=False)
+
+        def strip_path_prefix(path):
+            return os.path.relpath(path, os.environ.get("AIRFLOW_HOME"))
+
+        # prepend "(None,None)" to ensure that a test object is always created even if it's a no op.
+        return [(None, None)] + [
+            (strip_path_prefix(k), v.strip()) for k, v in dag_bag.import_errors.items()
+        ]
+
+
+def get_dags():
+    """
+    Generate a tuple of dag_id, <DAG objects> in the DagBag
+    """
+    with suppress_logging("airflow"):
+        dag_bag = DagBag(include_examples=False)
+
+    def strip_path_prefix(path):
+        return os.path.relpath(path, os.environ.get("AIRFLOW_HOME"))
+
+    return [(k, v, strip_path_prefix(v.fileloc)) for k, v in dag_bag.dags.items()]
+
+
+@pytest.mark.parametrize(
+    "rel_path,rv", get_import_errors(), ids=[x[0] for x in get_import_errors()]
+)
+def test_file_imports(rel_path, rv):
+    """Test for import errors on a file"""
+    if rel_path and rv:
+        raise Exception(f"{rel_path} failed to import with message \n {rv}")
+
+
+APPROVED_TAGS = {}
+
+
+@pytest.mark.parametrize(
+    "dag_id,dag,fileloc", get_dags(), ids=[x[2] for x in get_dags()]
+)
+def test_dag_tags(dag_id, dag, fileloc):
+    """
+    test if a DAG is tagged and if those TAGs are in the approved list
+    """
+    assert dag.tags, f"{dag_id} in {fileloc} has no tags"
+    if APPROVED_TAGS:
+        assert not set(dag.tags) - APPROVED_TAGS
+
+
+@pytest.mark.parametrize(
+    "dag_id,dag, fileloc", get_dags(), ids=[x[2] for x in get_dags()]
+)
+def test_dag_retries(dag_id, dag, fileloc):
+    """
+    test if a DAG has retries set
+    """
+    assert (
+        dag.default_args.get("retries", None) >= 2
+    ), f"{dag_id} in {fileloc} must have task retries >= 2."