Baraah A.

.dockerignore

@@ -0,0 +1,8 @@
+astro
+.git
+.env
+airflow_settings.yaml
+logs/
+.venv
+airflow.db
+airflow.cfg

.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
 {

.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)"

AI_ASSIST.md

@@ -1,12 +1,20 @@
 # 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:** 
+I received an Airflow task failure with the error:
+AirflowNotFoundException: The conn_id azure_pg isn't defined
 
-**What the model answered:** _Replace this section._
+I asked the LLM to explain the cause of this error and what steps were needed to fix the Airflow PostgreSQL connection.
 
-**What I kept, changed, or discarded, and why:** _Replace this section._
+
+**What the model answered:** 
+The model explained that the DAG was trying to use an Airflow connection named `azure_pg` through `PostgresHook`but this connection hadnt been created in the Airflow environment. It suggested checking the Airflow connections and adding the missing connection using the Astro CLI.
+
+**What I kept, changed, or discarded, and why:** 
+I kept the explanation that the issue was caused by a missing Airflow connection. I verified the solution by creating the `azure_pg` connection in my Astro Airflow environment and rerunning the DAG. The task succeeded afterward.
+I configured my own environment variables and connection settings.

ASSIGNMENT_REPORT.md

@@ -1,31 +1,49 @@
 # Assignment report
 
-<!-- Fill in every section below. Keep it short: a few sentences each. -->
+<!-- Replace every TODO. Keep it short: a few sentences per section. -->
 
 ## Schedule choice and reason
 
-_Replace this section._
+ we did a monthly schedule (@monthly) because the TLC taxi dataset is partitioned by month. Each DAG run processes one month of data based on Airflow logical_date.
 
 ## Task dependency graph
 
-_Replace: describe ingest -> dbt_run -> dbt_test and why order matters._
+describe the chain (ingest -> dbt_run -> dbt_test) and why the order matters.
+
+ingest_taxi_month -> dbt_run -> dbt_test
+
+The ingest task runs first because it downloads and loads the raw taxi data into PostgreSQL. After the raw data is available, 'dbt_run' transforms the data using the the class reference. Finally, 'dbt_test' validates the dbt models to ensure data quality.
+
+ingest loads raw data
+dbt transforms data
+dbt tests validate models.
 
 ## dbt project used
 
-_Replace: your Week 10 project or the class reference?_
+the class reference
 
 ## One debugging case I resolved
 
-_Replace: what failed, how you found the cause in the logs, and the fix._
+what failed, how you found the cause in the logs, and the fix.
+
+The first manual run failed because the DAG used the current date and tried downloading green_tripdata_2026-07.parquet, which was unavailable I think .
+
+I fixed this by triggering the DAG with a logical date (2024-01-01) and using Airflow logical_date instead of datetime.now().
 
-## Parameterized runs and backfill
+The DAG uses Airflow's logical date to derive the year-month partition.
 
-_Replace: how {{ ds }} / logical date drives the partition; the exact backfill create command you ran (with --max-active-runs 1)._
+The ingest task reads dag_run.logical_date and converts it to YYYY-MM format to build the TLC parquet URL.
 
-## Idempotency row counts (before / after re-run)
+Backfill command used:
+astro dev run backfill create \
+  --dag-id taxi_pipeline \
+  --from-date 2024-01-01 \
+  --to-date 2024-07-31 \
+  --max-active-runs 1
 
-_Replace: paste monthly counts before the re-run, then after. They must match._
+## shared Airflow deploy PR
+  https://github.com/lassebenni/c55-shared-airflow/pull/7
 
-## Shared Airflow deploy proof (if VM online)
 
-_Replace: merged c55-shared-airflow PR URL + path to your shared-UI screenshot in this repo._
+<!-- Target tier: also document your {{ ds }} parameter usage and the
+     backfill command(s) you ran, with before/after row counts. -->

RUNBOOK.md

@@ -1,22 +1,57 @@
 # 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 Astro Airflow environment:
 
-## How to run a backfill
-
-_Replace this section._
+```bash
+astro dev start
 
+## How to run a backfill
+The DAG uses a monthly schedule, so backfills should run month by month.
+
+astro dev run backfill create \
+  --dag-id taxi_pipeline \
+  --from-date 2024-01-01 \
+  --to-date 2024-07-31 \
+  --max-active-runs 1
+
+After the backfill finishes, verify the processed rows in PostgreSQL:
+
+SELECT
+    to_char(lpep_pickup_datetime, 'YYYY-MM') AS month,
+    COUNT(*) AS rows
+FROM airflow_baraah.raw_trips
+GROUP BY 1
+ORDER BY 1;
 ## How to inspect task logs
 
-_Replace this section._
+1. open the UI airflow 
+2. open taxi_pipeline
+3. slect the run DAG and then click the task 
+ingest_taxi_month
+dbt_run
+dbt_test
+4. open logs
 
 ## Top 3 likely failures and first response
 
-1. _Replace: symptom, first check, fix_
-2. _Replace this section._
-3. _Replace this section._
+1. TLC data download failure (HTTP 403 or missing parquet file)
+
+Symptom: ingest_taxi_month fails with an HTTP error.
+First check: Inspect the task log and verify the logical date used for the run.
+Fix: Use a valid historical logical date where the TLC parquet file exists (for example 2024-01-01).
+
+2. PostgreSQL connection failure
+Symptom: Error such as The conn_id azure_pg isn't defined.
+First check: Verify the Airflow connection exist.
+Fix: Add or update the PostgreSQL connection.
+
+3. dbt run or dbt test failure
+Symptom: dbt_run or dbt_test task becomes failed.
+First check: Open the task logs and check the dbt error message.
+Fix: Verify that the dbt project exists under include/dbt_project ,
+and that dbt is executed through: uvx --python 3.11

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()