Mareh 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,17 @@
 # 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:** "The dbt_run task failed with FATAL: password authentication failed for user. How do I fix this connection error in my Astro/Airflow project?"
 
-**What the model answered:** _Replace this section._
 
-**What I kept, changed, or discarded, and why:** _Replace this section._
+**What the model answered:** The model explained that dbt reads database credentials from environment variables mapped in profiles.yml. It provided instructions to check the .env file for missing PostgreSQL variables (PG_USER, PG_PASSWORD, PG_SCHEMA) and suggested restarting the Astro environment (astro dev stop && astro dev start) before clearing the task in the Airflow UI.
+
+**What I kept, changed, or discarded, and why:** Kept: The steps to add the missing PG_* variables into the .env file and restart Astro.
+
+Changed: Kept sensitive information (like my actual password and username) local without saving them in shared code files.
+
+Why: This directly solved the credential mismatch between Airflow and dbt while maintaining proper security practices for database credentials.

ASSIGNMENT_REPORT.md

@@ -1,31 +1,36 @@
-# Assignment report
+# Assignment Report
 
-<!-- Fill in every section below. Keep it short: a few sentences each. -->
+## Schedule Choice and Reason
 
-## Schedule choice and reason
+I chose a monthly schedule (`0 0 1 * *`) for the `taxi_pipeline` DAG. The dataset updates once a month, so running the pipeline on the first of every month keeps the data up to date without wasting resources.
 
-_Replace this section._
+## Task Dependency Graph
 
-## Task dependency graph
+The pipeline runs in a strict 3-step order:
+`ingest_taxi_month` $\rightarrow$ `dbt_run` $\rightarrow$ `dbt_test`.
 
-_Replace: describe ingest -> dbt_run -> dbt_test and why order matters._
+* **`ingest_taxi_month`** downloads the raw data into PostgreSQL first.
+* **`dbt_run`** transforms that raw data into clean dbt models.
+* **`dbt_test`** checks the cleaned data for quality errors.
 
-## dbt project used
+Order matters because dbt cannot transform or test data before it is ingested into the database.
 
-_Replace: your Week 10 project or the class reference?_
+## dbt Project Used
 
-## One debugging case I resolved
+I used the class reference dbt project, set up to build models inside my personal database schema.
 
-_Replace: what failed, how you found the cause in the logs, and the fix._
+## Debugging Case Resolved
 
-## Parameterized runs and backfill
+* **Problem:** The ingestion and dbt tasks failed with a database password authentication error.
+* **Diagnosis:** I checked the Airflow task logs and found that PostgreSQL rejected the connection due to incorrect credentials in the environment variables.
+* **Fix:** I updated `.env` with the correct database credentials and schema, restarted Astro, and cleared the failed tasks in the Airflow UI to rerun them successfully.
 
-_Replace: how {{ ds }} / logical date drives the partition; the exact backfill create command you ran (with --max-active-runs 1)._
+## Backfill Notes
 
-## Idempotency row counts (before / after re-run)
+* **Dynamic Dates:** The DAG uses the `{{ ds }}` parameter to automatically pick the correct date range for each run.
+* **Backfill Command:** I ran the backfill for 2024 using:
+```bash
+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 proof (if VM online)
-
-_Replace: merged c55-shared-airflow PR URL + path to your shared-UI screenshot in this repo._
+```
+* **Performance Note:** The backfill takes a long time to complete because `--max-active-runs 1` forces Airflow to process each month sequentially (one at a time) to avoid overloading the server.

RUNBOOK.md

@@ -1,22 +1,62 @@
 # 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
+Open Airflow in your web browser (http://localhost:8080).
 
-_Replace this section._
+Find taxi_pipeline in the list and make sure the toggle switch is turned ON.
 
+Click on taxi_pipeline.
+
+Click the Trigger button at the top right corner.
+
+Click Trigger DAG to start running it immediately.
 ## How to run a backfill
 
-_Replace this section._
+To process data for past months (for example, all months in 2024), open your terminal and run this command:
+
+astro dev run backfill create --dag-id taxi_pipeline --from-date 2024-01-01 --to-date 2024-07-31 --max-active-runs 1
+
+Note: --max-active-runs 1 forces Airflow to run one month at a time so it doesn't overload your computer or database.
 
 ## How to inspect task logs
 
-_Replace this section._
+Go to the taxi_pipeline page in Airflow.
+
+Click on the Grid tab.
+
+Click on the square box for the task you want to check (like ingest_taxi_month or dbt_run).
+
+On the right side panel that pops up, click Logs.
+
+Scroll down to the bottom of the text to see what went wrong.
 
 ## Top 3 likely failures and first response
 
-1. _Replace: symptom, first check, fix_
-2. _Replace this section._
-3. _Replace this section._
+1. Database Password / Connection Error
+
+Problem: Task turns red with a password authentication failed error message.
+
+What to check: Check your .env file to see if your password or username is wrong.
+
+How to fix: Fix your credentials in .env, run astro dev stop then astro dev start, and click Clear on the red task in Airflow to try again.
+
+2. Data Download Error
+
+Problem: ingest_taxi_month fails because it cannot fetch the dataset online.
+
+What to check: Check if your internet is working or if the file for that specific month actually exists.
+
+How to fix: Make sure you are connected to the internet, then click Clear on the task to rerun it.
+
+3. dbt Transformation Error
+
+Problem: dbt_run or dbt_test fails after the data was already downloaded.
+
+What to check: Open the task logs for dbt_run and look for SQL error messages at the bottom.
+
+How to fix: Fix the SQL code or test rules in your dbt project, save your files, and click Clear on the task in Airflow to test it again.
+
+

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