From c5f4934399de3cbde0b504ec32c2a0a84bff03c7 Mon Sep 17 00:00:00 2001 From: Lasse Benninga Date: Tue, 9 Jun 2026 17:29:42 +0200 Subject: [PATCH 01/20] scaffold: data-track/week-12/README.md --- data-track/week-12/README.md | 27 +++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) create mode 100644 data-track/week-12/README.md diff --git a/data-track/week-12/README.md b/data-track/week-12/README.md new file mode 100644 index 0000000..7269e14 --- /dev/null +++ b/data-track/week-12/README.md @@ -0,0 +1,27 @@ +# Week 12: Dashboarding with Metabase and Streamlit + +Practice exercises for [Week 12](https://www.notion.so/hackyourfuture/Week-12-Dashboarding-with-Metabase-and-Streamlit) of the Data Track. + +## Exercises + +| Exercise | Topic | +|---|---| +| [Exercise 1](exercise_1/) | Write a Metabase SQL Question | +| [Exercise 2](exercise_2/) | Build a Metabase dashboard with a date filter | +| [Exercise 3](exercise_3/) | Streamlit: last DAG run status panel | +| [Exercise 4](exercise_4/) | Streamlit: data-freshness panel from Postgres | + +## Prerequisites + +- Logged in to HYF Metabase (URL in Week 12, Ch2) +- Your Week 10 `fct_trips` table is populated in `dev_` on Azure Postgres +- Your Week 11 Airflow `ingest_taxi_month` DAG has at least one run + +## Running the Streamlit exercises + +```bash +cd exercise_3 # or exercise_4 +pip install -r requirements.txt +cp .env.example .env # fill in your credentials +streamlit run app.py +``` From 8950b1d0a8c8055db5926bb7c91f072d1d7d4eb7 Mon Sep 17 00:00:00 2001 From: Lasse Benninga Date: Tue, 9 Jun 2026 17:29:57 +0200 Subject: [PATCH 02/20] scaffold: data-track/week-12/exercise_1/README.md --- data-track/week-12/exercise_1/README.md | 41 +++++++++++++++++++++++++ 1 file changed, 41 insertions(+) create mode 100644 data-track/week-12/exercise_1/README.md diff --git a/data-track/week-12/exercise_1/README.md b/data-track/week-12/exercise_1/README.md new file mode 100644 index 0000000..1bc2844 --- /dev/null +++ b/data-track/week-12/exercise_1/README.md @@ -0,0 +1,41 @@ +# Exercise 1: Write a Metabase SQL Question + +Practice writing a SQL Question in Metabase using your `fct_trips` dbt mart table. + +## Prerequisites + +- Logged in to HYF Metabase +- Your `dev_.fct_trips` table is visible under Browse data + +## Task + +1. Log in to Metabase and go to **New → SQL query**. +2. Select the **Azure Postgres (HYF)** database. +3. Write a query that returns trip count by `pickup_borough`, ordered descending: + + ```sql + SELECT pickup_borough, + COUNT(*) AS trip_count + FROM dev_yourname.fct_trips + GROUP BY pickup_borough + ORDER BY trip_count DESC + ``` + + Replace `dev_yourname` with your actual schema name (e.g. `dev_jana`). + +4. Set the visualisation to **Bar chart**. +5. Save the Question to **Your Personal Collection** with the name "Trip count by borough". + +## Success criteria + +- The Question runs without error. +- The bar chart shows at least 3 boroughs with non-zero counts. +- The Question is saved (you can see it in your Personal Collection). + +## Stretch + +Write a second Question: average `fare_amount` by `pickup_borough`, filtered to exclude trips where `fare_amount = 0`. Save it as "Average fare by borough (non-zero)". + +## Solution + +See `solutions/README.md` for the reference SQL. From c574bfc8e876561fdd255ad629c3f6a54d4caa61 Mon Sep 17 00:00:00 2001 From: Lasse Benninga Date: Tue, 9 Jun 2026 17:29:59 +0200 Subject: [PATCH 03/20] scaffold: data-track/week-12/exercise_1/solutions/README.md --- .../week-12/exercise_1/solutions/README.md | 24 +++++++++++++++++++ 1 file changed, 24 insertions(+) create mode 100644 data-track/week-12/exercise_1/solutions/README.md diff --git a/data-track/week-12/exercise_1/solutions/README.md b/data-track/week-12/exercise_1/solutions/README.md new file mode 100644 index 0000000..da89f6a --- /dev/null +++ b/data-track/week-12/exercise_1/solutions/README.md @@ -0,0 +1,24 @@ +# Exercise 1 Solution + +## Minimum SQL + +```sql +SELECT pickup_borough, + COUNT(*) AS trip_count +FROM dev_yourname.fct_trips +GROUP BY pickup_borough +ORDER BY trip_count DESC +``` + +## Stretch SQL (average fare, excluding zero fares) + +```sql +SELECT pickup_borough, + AVG(fare_amount) AS avg_fare +FROM dev_yourname.fct_trips +WHERE fare_amount > 0 +GROUP BY pickup_borough +ORDER BY avg_fare DESC +``` + +Both queries should be saved as SQL Questions in your Personal Collection before proceeding to Exercise 2. From b6c6135feb39d49413bd6c985b757a960bb5384d Mon Sep 17 00:00:00 2001 From: Lasse Benninga Date: Tue, 9 Jun 2026 17:30:14 +0200 Subject: [PATCH 04/20] scaffold: data-track/week-12/exercise_2/README.md --- data-track/week-12/exercise_2/README.md | 41 +++++++++++++++++++++++++ 1 file changed, 41 insertions(+) create mode 100644 data-track/week-12/exercise_2/README.md diff --git a/data-track/week-12/exercise_2/README.md b/data-track/week-12/exercise_2/README.md new file mode 100644 index 0000000..110bf15 --- /dev/null +++ b/data-track/week-12/exercise_2/README.md @@ -0,0 +1,41 @@ +# Exercise 2: Build a Metabase Dashboard with a Date Filter + +Combine the Questions from Exercise 1 into a dashboard, then add a date filter. + +## Prerequisites + +- Exercise 1 completed: "Trip count by borough" Question saved in your Personal Collection +- At least one more Question saved (e.g. "Average fare by borough") + +## Task + +### Part A: Create the dashboard + +1. Go to **New → Dashboard**. +2. Name it "NYC Taxi Analytics: [Your Name]". +3. Click **Add a question** and add both Questions from Exercise 1. +4. Arrange the cards side by side using the drag handles. +5. Click **Done** to exit edit mode. + +### Part B: Add a date filter + +1. Re-enter edit mode (**Edit**). +2. Click **Filters → Date → Date Range**. +3. Open the filter settings for the "Trip count by borough" Question. +4. Map the filter to the `pickup_datetime` column. +5. Repeat for the "Average fare by borough" Question. +6. Click **Done** and test the filter: select a 30-day range and verify both panels update. + +## Success criteria + +- Dashboard contains at least 2 Questions. +- Date-range filter is connected to at least 2 Questions. +- Filtering to a specific month changes the trip counts shown. + +## Stretch + +Add a third Question: daily trip volume over time (line chart). Check that the date filter also connects to it. + +## Solution + +See `solutions/README.md` for step-by-step guidance. From 86b83a85cfe0123622b268524f990d1e8439d2e7 Mon Sep 17 00:00:00 2001 From: Lasse Benninga Date: Tue, 9 Jun 2026 17:30:16 +0200 Subject: [PATCH 05/20] scaffold: data-track/week-12/exercise_2/solutions/README.md --- data-track/week-12/exercise_2/solutions/README.md | 15 +++++++++++++++ 1 file changed, 15 insertions(+) create mode 100644 data-track/week-12/exercise_2/solutions/README.md diff --git a/data-track/week-12/exercise_2/solutions/README.md b/data-track/week-12/exercise_2/solutions/README.md new file mode 100644 index 0000000..6f61a8a --- /dev/null +++ b/data-track/week-12/exercise_2/solutions/README.md @@ -0,0 +1,15 @@ +# Exercise 2 Solution + +## Key steps + +1. Create dashboard via New → Dashboard. +2. Add Questions via the + button in edit mode. +3. Add filter: Edit → Filters → Date → Date Range. +4. Map filter: click the filter gear icon on each Question card → select the date column (`pickup_datetime`). +5. Done → test. + +## Common issue: filter has no effect + +If the filter does not change the results, the most likely cause is that you forgot to map the filter to the Question's date column. Each Question has its own mapping — adding the filter to the dashboard is not enough. + +Check: enter edit mode → click the filter label → confirm both Question cards show the mapped column. From ec4af0d59b48709d94165e5d75a4d753b9e2967a Mon Sep 17 00:00:00 2001 From: Lasse Benninga Date: Tue, 9 Jun 2026 17:30:41 +0200 Subject: [PATCH 06/20] scaffold: data-track/week-12/exercise_3/README.md --- data-track/week-12/exercise_3/README.md | 47 +++++++++++++++++++++++++ 1 file changed, 47 insertions(+) create mode 100644 data-track/week-12/exercise_3/README.md diff --git a/data-track/week-12/exercise_3/README.md b/data-track/week-12/exercise_3/README.md new file mode 100644 index 0000000..af71bf0 --- /dev/null +++ b/data-track/week-12/exercise_3/README.md @@ -0,0 +1,47 @@ +# Exercise 3: Streamlit Last DAG Run Status Panel + +Build a Streamlit app with one panel that calls the Airflow REST API and shows the last run state. + +## Prerequisites + +- Airflow URL, username, and password (from your teacher) +- Python 3.11+ and `pip` + +## Setup + +```bash +pip install -r requirements.txt +cp .env.example .env +# Fill in AIRFLOW_URL, AIRFLOW_USER, AIRFLOW_PASS in .env +streamlit run app.py +``` + +## Task + +Open `app.py`. Follow the `TODO` comments to: + +1. Implement `get_dag_runs(dag_id, limit)` using `requests.get` with Basic Auth. +2. Call it with `dag_id = "ingest_taxi_month"` and `limit = 1`. +3. Display the result as `st.success` (success), `st.error` (failed), or `st.warning` (other state). + +The app should look like this when working: + +```text +┌─────────────────────────────────────────┐ +│ Last run: success — started 2024-01-15 │ +└─────────────────────────────────────────┘ +``` + +## Success criteria + +- Running `streamlit run app.py` opens a page in the browser. +- The panel shows the real last run state of your `ingest_taxi_month` DAG. +- No credentials are hardcoded in `app.py`. + +## Stretch + +Add a `st.selectbox` in the sidebar to switch between two DAG IDs. The panel should update when you change the selection. + +## Solution + +See `solutions/app.py` for the reference implementation. From fdceafb6ae96bfe04c06bc0e5d6270994fee7876 Mon Sep 17 00:00:00 2001 From: Lasse Benninga Date: Tue, 9 Jun 2026 17:30:42 +0200 Subject: [PATCH 07/20] scaffold: data-track/week-12/exercise_3/app.py --- data-track/week-12/exercise_3/app.py | 53 ++++++++++++++++++++++++++++ 1 file changed, 53 insertions(+) create mode 100644 data-track/week-12/exercise_3/app.py diff --git a/data-track/week-12/exercise_3/app.py b/data-track/week-12/exercise_3/app.py new file mode 100644 index 0000000..f4c33ca --- /dev/null +++ b/data-track/week-12/exercise_3/app.py @@ -0,0 +1,53 @@ +""" +Exercise 3: Streamlit last DAG run status panel. +Fill in the TODOs to make the panel show real data. +""" + +import os + +import requests +import streamlit as st +from dotenv import load_dotenv + +load_dotenv() + +AIRFLOW_URL = os.environ["AIRFLOW_URL"] +AIRFLOW_USER = os.environ["AIRFLOW_USER"] +AIRFLOW_PASS = os.environ["AIRFLOW_PASS"] + + +@st.cache_data(ttl=60) +def get_dag_runs(dag_id: str, limit: int = 10) -> list: + """Return recent DAG runs from the Airflow REST API. + + TODO: implement this function. + Endpoint: GET {AIRFLOW_URL}/api/v1/dags/{dag_id}/dagRuns + Params: limit=limit, order_by="-execution_date" + Auth: requests.get(..., auth=(AIRFLOW_USER, AIRFLOW_PASS)) + Return: response.json()["dag_runs"] + """ + raise NotImplementedError("TODO: implement get_dag_runs") + + +st.title("Pipeline Health") +st.subheader("Last DAG run") + +dag_id = "ingest_taxi_month" + +try: + runs = get_dag_runs(dag_id, limit=1) + if runs: + last = runs[0] + state = last["state"] + if state == "success": + st.success(f"Last run: **{state}** — started {last['start_date']}") + elif state == "failed": + st.error(f"Last run: **{state}** — check Airflow logs") + else: + st.warning(f"Last run: **{state}**") + else: + st.info("No runs found.") +except NotImplementedError: + st.warning("Implement `get_dag_runs` to see live data.") +except Exception as exc: + st.error(f"Error: {exc}") From 0932094662d68b5142d5388241a8e3326aa8f398 Mon Sep 17 00:00:00 2001 From: Lasse Benninga Date: Tue, 9 Jun 2026 17:30:44 +0200 Subject: [PATCH 08/20] scaffold: data-track/week-12/exercise_3/requirements.txt --- data-track/week-12/exercise_3/requirements.txt | 3 +++ 1 file changed, 3 insertions(+) create mode 100644 data-track/week-12/exercise_3/requirements.txt diff --git a/data-track/week-12/exercise_3/requirements.txt b/data-track/week-12/exercise_3/requirements.txt new file mode 100644 index 0000000..add1236 --- /dev/null +++ b/data-track/week-12/exercise_3/requirements.txt @@ -0,0 +1,3 @@ +streamlit>=1.35.0 +requests>=2.31.0 +python-dotenv>=1.0.0 From f21dfe7b09b78c9b13ff7cf6bd6c10fc5a35edda Mon Sep 17 00:00:00 2001 From: Lasse Benninga Date: Tue, 9 Jun 2026 17:30:45 +0200 Subject: [PATCH 09/20] scaffold: data-track/week-12/exercise_3/.env.example --- data-track/week-12/exercise_3/.env.example | 3 +++ 1 file changed, 3 insertions(+) create mode 100644 data-track/week-12/exercise_3/.env.example diff --git a/data-track/week-12/exercise_3/.env.example b/data-track/week-12/exercise_3/.env.example new file mode 100644 index 0000000..278f575 --- /dev/null +++ b/data-track/week-12/exercise_3/.env.example @@ -0,0 +1,3 @@ +AIRFLOW_URL=https://your-airflow.example.com +AIRFLOW_USER=admin +AIRFLOW_PASS=your-password From 922eac5eb484aa658fcef03d8c7746275c07be6f Mon Sep 17 00:00:00 2001 From: Lasse Benninga Date: Tue, 9 Jun 2026 17:30:47 +0200 Subject: [PATCH 10/20] scaffold: data-track/week-12/exercise_3/solutions/app.py --- .../week-12/exercise_3/solutions/app.py | 45 +++++++++++++++++++ 1 file changed, 45 insertions(+) create mode 100644 data-track/week-12/exercise_3/solutions/app.py diff --git a/data-track/week-12/exercise_3/solutions/app.py b/data-track/week-12/exercise_3/solutions/app.py new file mode 100644 index 0000000..b5993c0 --- /dev/null +++ b/data-track/week-12/exercise_3/solutions/app.py @@ -0,0 +1,45 @@ +"""Exercise 3 solution.""" + +import os + +import requests +import streamlit as st +from dotenv import load_dotenv + +load_dotenv() + +AIRFLOW_URL = os.environ["AIRFLOW_URL"] +AIRFLOW_USER = os.environ["AIRFLOW_USER"] +AIRFLOW_PASS = os.environ["AIRFLOW_PASS"] + + +@st.cache_data(ttl=60) +def get_dag_runs(dag_id: str, limit: int = 10) -> list: + url = f"{AIRFLOW_URL}/api/v1/dags/{dag_id}/dagRuns" + r = requests.get( + url, + params={"limit": limit, "order_by": "-execution_date"}, + auth=(AIRFLOW_USER, AIRFLOW_PASS), + timeout=10, + ) + r.raise_for_status() + return r.json()["dag_runs"] + + +st.title("Pipeline Health") +st.subheader("Last DAG run") + +dag_id = "ingest_taxi_month" + +runs = get_dag_runs(dag_id, limit=1) +if runs: + last = runs[0] + state = last["state"] + if state == "success": + st.success(f"Last run: **{state}** — started {last['start_date']}") + elif state == "failed": + st.error(f"Last run: **{state}** — check Airflow logs") + else: + st.warning(f"Last run: **{state}**") +else: + st.info("No runs found.") From f5f3a198cd2f5d6cd5623f5a58a945220e6f59c1 Mon Sep 17 00:00:00 2001 From: Lasse Benninga Date: Tue, 9 Jun 2026 17:31:17 +0200 Subject: [PATCH 11/20] scaffold: data-track/week-12/exercise_4/README.md --- data-track/week-12/exercise_4/README.md | 39 +++++++++++++++++++++++++ 1 file changed, 39 insertions(+) create mode 100644 data-track/week-12/exercise_4/README.md diff --git a/data-track/week-12/exercise_4/README.md b/data-track/week-12/exercise_4/README.md new file mode 100644 index 0000000..9695029 --- /dev/null +++ b/data-track/week-12/exercise_4/README.md @@ -0,0 +1,39 @@ +# Exercise 4: Streamlit Data-Freshness Panel + +Extend the app from Exercise 3 with a second panel that reads `fct_trips` row count and last-updated timestamp from Azure Postgres. + +## Prerequisites + +- Exercise 3 working (last run panel showing real data) +- Azure Postgres connection string (`PG_URL`) from your teacher + +## Setup + +```bash +pip install -r requirements.txt +cp .env.example .env +# Add PG_URL to .env (in addition to the Airflow env vars) +streamlit run app.py +``` + +## Task + +Open `app.py`. Follow the `TODO` comments to: + +1. Implement `get_fct_trips_freshness(pg_url)` using `psycopg2`. +2. Query `MAX(pickup_datetime)` and `COUNT(*)` from `dev_.fct_trips`. +3. Display as two `st.metric` widgets: "Row count" and "Last pickup". + +## Success criteria + +- Both `st.metric` widgets show non-zero / non-null values from real data. +- The function uses `@st.cache_data(ttl=60)`. +- Credentials come from environment variables only. + +## Stretch + +Calculate how many hours ago the last pickup was and display it as the `delta` parameter of `st.metric`. For example: "Last pickup: 2024-01-15 03:00 UTC — 14 h ago". + +## Solution + +See `solutions/app.py`. From cd6e35dfdec81468d80043105e613b0d00a022a4 Mon Sep 17 00:00:00 2001 From: Lasse Benninga Date: Tue, 9 Jun 2026 17:31:18 +0200 Subject: [PATCH 12/20] scaffold: data-track/week-12/exercise_4/app.py --- data-track/week-12/exercise_4/app.py | 77 ++++++++++++++++++++++++++++ 1 file changed, 77 insertions(+) create mode 100644 data-track/week-12/exercise_4/app.py diff --git a/data-track/week-12/exercise_4/app.py b/data-track/week-12/exercise_4/app.py new file mode 100644 index 0000000..95ab27d --- /dev/null +++ b/data-track/week-12/exercise_4/app.py @@ -0,0 +1,77 @@ +""" +Exercise 4: add a data-freshness panel to the pipeline health app. +""" + +import os +from datetime import datetime, timezone + +import psycopg2 +import requests +import streamlit as st +from dotenv import load_dotenv + +load_dotenv() + +AIRFLOW_URL = os.environ["AIRFLOW_URL"] +AIRFLOW_USER = os.environ["AIRFLOW_USER"] +AIRFLOW_PASS = os.environ["AIRFLOW_PASS"] +PG_URL = os.environ["PG_URL"] +PG_SCHEMA = os.environ.get("PG_SCHEMA", "dev_yourname") + + +@st.cache_data(ttl=60) +def get_dag_runs(dag_id: str, limit: int = 10) -> list: + url = f"{AIRFLOW_URL}/api/v1/dags/{dag_id}/dagRuns" + r = requests.get( + url, + params={"limit": limit, "order_by": "-execution_date"}, + auth=(AIRFLOW_USER, AIRFLOW_PASS), + timeout=10, + ) + r.raise_for_status() + return r.json()["dag_runs"] + + +@st.cache_data(ttl=60) +def get_fct_trips_freshness(pg_url: str, schema: str) -> dict: + """Return row count and last pickup_datetime for fct_trips. + + TODO: implement this function. + Use psycopg2.connect(pg_url) to connect. + Run: SELECT COUNT(*), MAX(pickup_datetime) FROM {schema}.fct_trips + Return a dict: {"row_count": int, "last_pickup": datetime or None} + """ + raise NotImplementedError("TODO: implement get_fct_trips_freshness") + + +st.title("Pipeline Health") + +# Panel 1: last run +st.subheader("Last DAG run") +dag_id = "ingest_taxi_month" +try: + runs = get_dag_runs(dag_id, limit=1) + if runs: + last = runs[0] + state = last["state"] + if state == "success": + st.success(f"Last run: **{state}** — started {last['start_date']}") + elif state == "failed": + st.error(f"Last run: **{state}** — check Airflow logs") + else: + st.warning(f"Last run: **{state}**") +except Exception as exc: + st.error(f"Airflow error: {exc}") + +# Panel 2: data freshness +st.subheader("Data freshness") +try: + freshness = get_fct_trips_freshness(PG_URL, PG_SCHEMA) + col1, col2 = st.columns(2) + col1.metric("Row count", f"{freshness['row_count']:,}") + last = freshness["last_pickup"] + col2.metric("Last pickup", str(last) if last else "None") +except NotImplementedError: + st.warning("Implement `get_fct_trips_freshness` to see freshness data.") +except Exception as exc: + st.error(f"Postgres error: {exc}") From 08daa82a9f4fa6ad23f8c92bc6033eb280ffdbac Mon Sep 17 00:00:00 2001 From: Lasse Benninga Date: Tue, 9 Jun 2026 17:31:20 +0200 Subject: [PATCH 13/20] scaffold: data-track/week-12/exercise_4/requirements.txt --- data-track/week-12/exercise_4/requirements.txt | 4 ++++ 1 file changed, 4 insertions(+) create mode 100644 data-track/week-12/exercise_4/requirements.txt diff --git a/data-track/week-12/exercise_4/requirements.txt b/data-track/week-12/exercise_4/requirements.txt new file mode 100644 index 0000000..2cf1fd3 --- /dev/null +++ b/data-track/week-12/exercise_4/requirements.txt @@ -0,0 +1,4 @@ +streamlit>=1.35.0 +requests>=2.31.0 +psycopg2-binary>=2.9.9 +python-dotenv>=1.0.0 From 9a2ef1b957b15da184d403dced16822933a614fd Mon Sep 17 00:00:00 2001 From: Lasse Benninga Date: Tue, 9 Jun 2026 17:31:21 +0200 Subject: [PATCH 14/20] scaffold: data-track/week-12/exercise_4/.env.example --- data-track/week-12/exercise_4/.env.example | 8 ++++++++ 1 file changed, 8 insertions(+) create mode 100644 data-track/week-12/exercise_4/.env.example diff --git a/data-track/week-12/exercise_4/.env.example b/data-track/week-12/exercise_4/.env.example new file mode 100644 index 0000000..b92196f --- /dev/null +++ b/data-track/week-12/exercise_4/.env.example @@ -0,0 +1,8 @@ +AIRFLOW_URL=https://your-airflow.example.com +AIRFLOW_USER=admin +AIRFLOW_PASS=your-password + +# PostgreSQL connection string +PG_URL=postgresql://pipeline_user:your-pg-password@your-pg-host:5432/team1 +# Your dev schema name (e.g. dev_jana) +PG_SCHEMA=dev_yourname From 04ce2e037420e8e92a630baa06904cbc6ec4b3e5 Mon Sep 17 00:00:00 2001 From: Lasse Benninga Date: Tue, 9 Jun 2026 17:31:23 +0200 Subject: [PATCH 15/20] scaffold: data-track/week-12/exercise_4/solutions/app.py --- .../week-12/exercise_4/solutions/app.py | 62 +++++++++++++++++++ 1 file changed, 62 insertions(+) create mode 100644 data-track/week-12/exercise_4/solutions/app.py diff --git a/data-track/week-12/exercise_4/solutions/app.py b/data-track/week-12/exercise_4/solutions/app.py new file mode 100644 index 0000000..cbe26d1 --- /dev/null +++ b/data-track/week-12/exercise_4/solutions/app.py @@ -0,0 +1,62 @@ +"""Exercise 4 solution.""" + +import os + +import psycopg2 +import requests +import streamlit as st +from dotenv import load_dotenv + +load_dotenv() + +AIRFLOW_URL = os.environ["AIRFLOW_URL"] +AIRFLOW_USER = os.environ["AIRFLOW_USER"] +AIRFLOW_PASS = os.environ["AIRFLOW_PASS"] +PG_URL = os.environ["PG_URL"] +PG_SCHEMA = os.environ.get("PG_SCHEMA", "dev_yourname") + + +@st.cache_data(ttl=60) +def get_dag_runs(dag_id: str, limit: int = 10) -> list: + url = f"{AIRFLOW_URL}/api/v1/dags/{dag_id}/dagRuns" + r = requests.get( + url, + params={"limit": limit, "order_by": "-execution_date"}, + auth=(AIRFLOW_USER, AIRFLOW_PASS), + timeout=10, + ) + r.raise_for_status() + return r.json()["dag_runs"] + + +@st.cache_data(ttl=60) +def get_fct_trips_freshness(pg_url: str, schema: str) -> dict: + with psycopg2.connect(pg_url) as conn, conn.cursor() as cur: + cur.execute( + f"SELECT COUNT(*), MAX(pickup_datetime) FROM {schema}.fct_trips" + ) + row_count, last_pickup = cur.fetchone() + return {"row_count": row_count or 0, "last_pickup": last_pickup} + + +st.title("Pipeline Health") + +st.subheader("Last DAG run") +dag_id = "ingest_taxi_month" +runs = get_dag_runs(dag_id, limit=1) +if runs: + last = runs[0] + state = last["state"] + if state == "success": + st.success(f"Last run: **{state}** — started {last['start_date']}") + elif state == "failed": + st.error(f"Last run: **{state}** — check Airflow logs") + else: + st.warning(f"Last run: **{state}**") + +st.subheader("Data freshness") +freshness = get_fct_trips_freshness(PG_URL, PG_SCHEMA) +col1, col2 = st.columns(2) +col1.metric("Row count", f"{freshness['row_count']:,}") +last = freshness["last_pickup"] +col2.metric("Last pickup", str(last) if last else "None") From df2334b3e36990d20ba77ae0727a771340ccf8de Mon Sep 17 00:00:00 2001 From: Lasse Benninga Date: Thu, 11 Jun 2026 00:25:02 +0200 Subject: [PATCH 16/20] fix(week-12): Airflow 3 REST API (/api/v2 + logical_date) in practice exercises The class shared Airflow VM runs Airflow 3. The Streamlit exercises used the Airflow 2 pattern (/api/v1 + order_by=-execution_date), which 404s against the v3 API. Corrected ex3/ex4 starters and solutions to /api/v2 and -logical_date, matching the Week 12 chapters. Also removed em dashes from the exercise files (project style: colons/commas, never em dashes). Co-Authored-By: Claude Opus 4.8 (1M context) --- data-track/week-12/exercise_2/solutions/README.md | 2 +- data-track/week-12/exercise_3/README.md | 2 +- data-track/week-12/exercise_3/app.py | 8 ++++---- data-track/week-12/exercise_3/solutions/app.py | 8 ++++---- data-track/week-12/exercise_4/README.md | 2 +- data-track/week-12/exercise_4/app.py | 8 ++++---- data-track/week-12/exercise_4/solutions/app.py | 8 ++++---- 7 files changed, 19 insertions(+), 19 deletions(-) diff --git a/data-track/week-12/exercise_2/solutions/README.md b/data-track/week-12/exercise_2/solutions/README.md index 6f61a8a..3c6a38d 100644 --- a/data-track/week-12/exercise_2/solutions/README.md +++ b/data-track/week-12/exercise_2/solutions/README.md @@ -10,6 +10,6 @@ ## Common issue: filter has no effect -If the filter does not change the results, the most likely cause is that you forgot to map the filter to the Question's date column. Each Question has its own mapping — adding the filter to the dashboard is not enough. +If the filter does not change the results, the most likely cause is that you forgot to map the filter to the Question's date column. Each Question has its own mapping, adding the filter to the dashboard is not enough. Check: enter edit mode → click the filter label → confirm both Question cards show the mapped column. diff --git a/data-track/week-12/exercise_3/README.md b/data-track/week-12/exercise_3/README.md index af71bf0..a9eb86c 100644 --- a/data-track/week-12/exercise_3/README.md +++ b/data-track/week-12/exercise_3/README.md @@ -28,7 +28,7 @@ The app should look like this when working: ```text ┌─────────────────────────────────────────┐ -│ Last run: success — started 2024-01-15 │ +│ Last run: success - started 2024-01-15 │ └─────────────────────────────────────────┘ ``` diff --git a/data-track/week-12/exercise_3/app.py b/data-track/week-12/exercise_3/app.py index f4c33ca..7ee23bf 100644 --- a/data-track/week-12/exercise_3/app.py +++ b/data-track/week-12/exercise_3/app.py @@ -21,8 +21,8 @@ def get_dag_runs(dag_id: str, limit: int = 10) -> list: """Return recent DAG runs from the Airflow REST API. TODO: implement this function. - Endpoint: GET {AIRFLOW_URL}/api/v1/dags/{dag_id}/dagRuns - Params: limit=limit, order_by="-execution_date" + Endpoint: GET {AIRFLOW_URL}/api/v2/dags/{dag_id}/dagRuns + Params: limit=limit, order_by="-logical_date" Auth: requests.get(..., auth=(AIRFLOW_USER, AIRFLOW_PASS)) Return: response.json()["dag_runs"] """ @@ -40,9 +40,9 @@ def get_dag_runs(dag_id: str, limit: int = 10) -> list: last = runs[0] state = last["state"] if state == "success": - st.success(f"Last run: **{state}** — started {last['start_date']}") + st.success(f"Last run: **{state}**, started {last['start_date']}") elif state == "failed": - st.error(f"Last run: **{state}** — check Airflow logs") + st.error(f"Last run: **{state}**, check Airflow logs") else: st.warning(f"Last run: **{state}**") else: diff --git a/data-track/week-12/exercise_3/solutions/app.py b/data-track/week-12/exercise_3/solutions/app.py index b5993c0..5434900 100644 --- a/data-track/week-12/exercise_3/solutions/app.py +++ b/data-track/week-12/exercise_3/solutions/app.py @@ -15,10 +15,10 @@ @st.cache_data(ttl=60) def get_dag_runs(dag_id: str, limit: int = 10) -> list: - url = f"{AIRFLOW_URL}/api/v1/dags/{dag_id}/dagRuns" + url = f"{AIRFLOW_URL}/api/v2/dags/{dag_id}/dagRuns" r = requests.get( url, - params={"limit": limit, "order_by": "-execution_date"}, + params={"limit": limit, "order_by": "-logical_date"}, auth=(AIRFLOW_USER, AIRFLOW_PASS), timeout=10, ) @@ -36,9 +36,9 @@ def get_dag_runs(dag_id: str, limit: int = 10) -> list: last = runs[0] state = last["state"] if state == "success": - st.success(f"Last run: **{state}** — started {last['start_date']}") + st.success(f"Last run: **{state}**, started {last['start_date']}") elif state == "failed": - st.error(f"Last run: **{state}** — check Airflow logs") + st.error(f"Last run: **{state}**, check Airflow logs") else: st.warning(f"Last run: **{state}**") else: diff --git a/data-track/week-12/exercise_4/README.md b/data-track/week-12/exercise_4/README.md index 9695029..5783d3e 100644 --- a/data-track/week-12/exercise_4/README.md +++ b/data-track/week-12/exercise_4/README.md @@ -32,7 +32,7 @@ Open `app.py`. Follow the `TODO` comments to: ## Stretch -Calculate how many hours ago the last pickup was and display it as the `delta` parameter of `st.metric`. For example: "Last pickup: 2024-01-15 03:00 UTC — 14 h ago". +Calculate how many hours ago the last pickup was and display it as the `delta` parameter of `st.metric`. For example: "Last pickup: 2024-01-15 03:00 UTC, 14 h ago". ## Solution diff --git a/data-track/week-12/exercise_4/app.py b/data-track/week-12/exercise_4/app.py index 95ab27d..d63b8f1 100644 --- a/data-track/week-12/exercise_4/app.py +++ b/data-track/week-12/exercise_4/app.py @@ -21,10 +21,10 @@ @st.cache_data(ttl=60) def get_dag_runs(dag_id: str, limit: int = 10) -> list: - url = f"{AIRFLOW_URL}/api/v1/dags/{dag_id}/dagRuns" + url = f"{AIRFLOW_URL}/api/v2/dags/{dag_id}/dagRuns" r = requests.get( url, - params={"limit": limit, "order_by": "-execution_date"}, + params={"limit": limit, "order_by": "-logical_date"}, auth=(AIRFLOW_USER, AIRFLOW_PASS), timeout=10, ) @@ -55,9 +55,9 @@ def get_fct_trips_freshness(pg_url: str, schema: str) -> dict: last = runs[0] state = last["state"] if state == "success": - st.success(f"Last run: **{state}** — started {last['start_date']}") + st.success(f"Last run: **{state}**, started {last['start_date']}") elif state == "failed": - st.error(f"Last run: **{state}** — check Airflow logs") + st.error(f"Last run: **{state}**, check Airflow logs") else: st.warning(f"Last run: **{state}**") except Exception as exc: diff --git a/data-track/week-12/exercise_4/solutions/app.py b/data-track/week-12/exercise_4/solutions/app.py index cbe26d1..d4c5efc 100644 --- a/data-track/week-12/exercise_4/solutions/app.py +++ b/data-track/week-12/exercise_4/solutions/app.py @@ -18,10 +18,10 @@ @st.cache_data(ttl=60) def get_dag_runs(dag_id: str, limit: int = 10) -> list: - url = f"{AIRFLOW_URL}/api/v1/dags/{dag_id}/dagRuns" + url = f"{AIRFLOW_URL}/api/v2/dags/{dag_id}/dagRuns" r = requests.get( url, - params={"limit": limit, "order_by": "-execution_date"}, + params={"limit": limit, "order_by": "-logical_date"}, auth=(AIRFLOW_USER, AIRFLOW_PASS), timeout=10, ) @@ -48,9 +48,9 @@ def get_fct_trips_freshness(pg_url: str, schema: str) -> dict: last = runs[0] state = last["state"] if state == "success": - st.success(f"Last run: **{state}** — started {last['start_date']}") + st.success(f"Last run: **{state}**, started {last['start_date']}") elif state == "failed": - st.error(f"Last run: **{state}** — check Airflow logs") + st.error(f"Last run: **{state}**, check Airflow logs") else: st.warning(f"Last run: **{state}**") From 90f9517da434916e7ef086ed68ef00dbad98e44e Mon Sep 17 00:00:00 2001 From: Lasse Benninga Date: Thu, 11 Jun 2026 15:59:51 +0200 Subject: [PATCH 17/20] fix(week-12): Airflow 3 token auth in Streamlit exercises (not Basic Auth) Live smoke-test of the shared Airflow 3.2 VM: auth=(user,pass) returns 401 on /api/v2; the API needs a JWT from POST /auth/token sent as a Bearer header. Added a cached get_airflow_token() helper to ex3/ex4 (starter + solution) and switched every dagRuns call to Authorization: Bearer. Updated the ex3 starter TODO to describe the token flow. Co-Authored-By: Claude Opus 4.8 (1M context) --- data-track/week-12/exercise_3/app.py | 4 +++- data-track/week-12/exercise_3/solutions/app.py | 12 +++++++++++- data-track/week-12/exercise_4/app.py | 12 +++++++++++- data-track/week-12/exercise_4/solutions/app.py | 12 +++++++++++- 4 files changed, 36 insertions(+), 4 deletions(-) diff --git a/data-track/week-12/exercise_3/app.py b/data-track/week-12/exercise_3/app.py index 7ee23bf..fbe23f0 100644 --- a/data-track/week-12/exercise_3/app.py +++ b/data-track/week-12/exercise_3/app.py @@ -23,7 +23,9 @@ def get_dag_runs(dag_id: str, limit: int = 10) -> list: TODO: implement this function. Endpoint: GET {AIRFLOW_URL}/api/v2/dags/{dag_id}/dagRuns Params: limit=limit, order_by="-logical_date" - Auth: requests.get(..., auth=(AIRFLOW_USER, AIRFLOW_PASS)) + Auth: Airflow 3 uses a token, not Basic Auth. First POST to + {AIRFLOW_URL}/auth/token with {"username":..., "password":...} to get + an access_token, then send headers={"Authorization": f"Bearer {token}"}. Return: response.json()["dag_runs"] """ raise NotImplementedError("TODO: implement get_dag_runs") diff --git a/data-track/week-12/exercise_3/solutions/app.py b/data-track/week-12/exercise_3/solutions/app.py index 5434900..b44ccdf 100644 --- a/data-track/week-12/exercise_3/solutions/app.py +++ b/data-track/week-12/exercise_3/solutions/app.py @@ -13,13 +13,23 @@ AIRFLOW_PASS = os.environ["AIRFLOW_PASS"] +@st.cache_data(ttl=3000) # Airflow 3 uses token auth; the token is valid for 24h +def get_airflow_token() -> str: + r = requests.post( + f"{AIRFLOW_URL}/auth/token", + json={"username": AIRFLOW_USER, "password": AIRFLOW_PASS}, + ) + r.raise_for_status() + return r.json()["access_token"] + + @st.cache_data(ttl=60) def get_dag_runs(dag_id: str, limit: int = 10) -> list: url = f"{AIRFLOW_URL}/api/v2/dags/{dag_id}/dagRuns" r = requests.get( url, params={"limit": limit, "order_by": "-logical_date"}, - auth=(AIRFLOW_USER, AIRFLOW_PASS), + headers={"Authorization": f"Bearer {get_airflow_token()}"}, timeout=10, ) r.raise_for_status() diff --git a/data-track/week-12/exercise_4/app.py b/data-track/week-12/exercise_4/app.py index d63b8f1..e814a65 100644 --- a/data-track/week-12/exercise_4/app.py +++ b/data-track/week-12/exercise_4/app.py @@ -15,6 +15,16 @@ AIRFLOW_URL = os.environ["AIRFLOW_URL"] AIRFLOW_USER = os.environ["AIRFLOW_USER"] AIRFLOW_PASS = os.environ["AIRFLOW_PASS"] + + +@st.cache_data(ttl=3000) # Airflow 3 uses token auth; the token is valid for 24h +def get_airflow_token() -> str: + r = requests.post( + f"{AIRFLOW_URL}/auth/token", + json={"username": AIRFLOW_USER, "password": AIRFLOW_PASS}, + ) + r.raise_for_status() + return r.json()["access_token"] PG_URL = os.environ["PG_URL"] PG_SCHEMA = os.environ.get("PG_SCHEMA", "dev_yourname") @@ -25,7 +35,7 @@ def get_dag_runs(dag_id: str, limit: int = 10) -> list: r = requests.get( url, params={"limit": limit, "order_by": "-logical_date"}, - auth=(AIRFLOW_USER, AIRFLOW_PASS), + headers={"Authorization": f"Bearer {get_airflow_token()}"}, timeout=10, ) r.raise_for_status() diff --git a/data-track/week-12/exercise_4/solutions/app.py b/data-track/week-12/exercise_4/solutions/app.py index d4c5efc..16a2c05 100644 --- a/data-track/week-12/exercise_4/solutions/app.py +++ b/data-track/week-12/exercise_4/solutions/app.py @@ -12,6 +12,16 @@ AIRFLOW_URL = os.environ["AIRFLOW_URL"] AIRFLOW_USER = os.environ["AIRFLOW_USER"] AIRFLOW_PASS = os.environ["AIRFLOW_PASS"] + + +@st.cache_data(ttl=3000) # Airflow 3 uses token auth; the token is valid for 24h +def get_airflow_token() -> str: + r = requests.post( + f"{AIRFLOW_URL}/auth/token", + json={"username": AIRFLOW_USER, "password": AIRFLOW_PASS}, + ) + r.raise_for_status() + return r.json()["access_token"] PG_URL = os.environ["PG_URL"] PG_SCHEMA = os.environ.get("PG_SCHEMA", "dev_yourname") @@ -22,7 +32,7 @@ def get_dag_runs(dag_id: str, limit: int = 10) -> list: r = requests.get( url, params={"limit": limit, "order_by": "-logical_date"}, - auth=(AIRFLOW_USER, AIRFLOW_PASS), + headers={"Authorization": f"Bearer {get_airflow_token()}"}, timeout=10, ) r.raise_for_status() From 47cd00fdbeb1a854b1d0249ef4a5ccf03c77a13e Mon Sep 17 00:00:00 2001 From: Lasse Benninga Date: Thu, 11 Jun 2026 16:00:03 +0200 Subject: [PATCH 18/20] fix(week-12): ex3 README describes the token flow, not Basic Auth Co-Authored-By: Claude Opus 4.8 (1M context) --- data-track/week-12/exercise_3/README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/data-track/week-12/exercise_3/README.md b/data-track/week-12/exercise_3/README.md index a9eb86c..18ae3bd 100644 --- a/data-track/week-12/exercise_3/README.md +++ b/data-track/week-12/exercise_3/README.md @@ -20,7 +20,7 @@ streamlit run app.py Open `app.py`. Follow the `TODO` comments to: -1. Implement `get_dag_runs(dag_id, limit)` using `requests.get` with Basic Auth. +1. Implement `get_dag_runs(dag_id, limit)`: Airflow 3 uses token auth, so first `POST` to `/auth/token` for a JWT, then call `/api/v2/.../dagRuns` with an `Authorization: Bearer ` header. 2. Call it with `dag_id = "ingest_taxi_month"` and `limit = 1`. 3. Display the result as `st.success` (success), `st.error` (failed), or `st.warning` (other state). From f532f20b6c6c2043d736dc9d1624e80cc51a557d Mon Sep 17 00:00:00 2001 From: Lasse Benninga Date: Tue, 7 Jul 2026 14:36:56 +0200 Subject: [PATCH 19/20] =?UTF-8?q?chore(data-track):=20flip=20weeks=2011/12?= =?UTF-8?q?=20=E2=80=94=20dashboarding=20becomes=20week=2011=20(Airflow-fr?= =?UTF-8?q?ee)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Move the dashboarding week from week-12 to week-11 to staff each week to a teacher's strength (the Airflow expert is only available in week 12). - Rename data-track/week-12 -> data-track/week-11 (history preserved). - README: drop the Airflow DAG prerequisite, add PG_URL, TODO on Notion link. - Streamlit ex3: rewrite from "last DAG run" panel to KPI tiles from fct_trips. - Streamlit ex4: rewrite from Airflow panel to daily trip-volume chart + existing Postgres freshness panel. - Strip Airflow env vars and the requests dependency; add pandas for ex4. - Metabase ex1/ex2 unchanged (already dbt/Postgres only). Out of scope: authoring the new week-12 Airflow exercises and editing the Notion chapter prose/slug. Co-Authored-By: Claude Opus 4.8 (1M context) --- data-track/week-11/README.md | 32 +++++++ .../{week-12 => week-11}/exercise_1/README.md | 0 .../exercise_1/solutions/README.md | 0 .../{week-12 => week-11}/exercise_2/README.md | 0 .../exercise_2/solutions/README.md | 0 .../exercise_3}/.env.example | 4 - data-track/week-11/exercise_3/README.md | 54 ++++++++++++ data-track/week-11/exercise_3/app.py | 42 +++++++++ .../exercise_3}/requirements.txt | 1 - .../week-11/exercise_3/solutions/app.py | 36 ++++++++ data-track/week-11/exercise_4/.env.example | 4 + data-track/week-11/exercise_4/README.md | 47 ++++++++++ data-track/week-11/exercise_4/app.py | 65 ++++++++++++++ .../exercise_4}/requirements.txt | 3 +- .../week-11/exercise_4/solutions/app.py | 48 ++++++++++ data-track/week-12/README.md | 27 ------ data-track/week-12/exercise_3/.env.example | 3 - data-track/week-12/exercise_3/README.md | 47 ---------- data-track/week-12/exercise_3/app.py | 55 ------------ .../week-12/exercise_3/solutions/app.py | 55 ------------ data-track/week-12/exercise_4/README.md | 39 --------- data-track/week-12/exercise_4/app.py | 87 ------------------- .../week-12/exercise_4/solutions/app.py | 72 --------------- 23 files changed, 330 insertions(+), 391 deletions(-) create mode 100644 data-track/week-11/README.md rename data-track/{week-12 => week-11}/exercise_1/README.md (100%) rename data-track/{week-12 => week-11}/exercise_1/solutions/README.md (100%) rename data-track/{week-12 => week-11}/exercise_2/README.md (100%) rename data-track/{week-12 => week-11}/exercise_2/solutions/README.md (100%) rename data-track/{week-12/exercise_4 => week-11/exercise_3}/.env.example (64%) create mode 100644 data-track/week-11/exercise_3/README.md create mode 100644 data-track/week-11/exercise_3/app.py rename data-track/{week-12/exercise_4 => week-11/exercise_3}/requirements.txt (78%) create mode 100644 data-track/week-11/exercise_3/solutions/app.py create mode 100644 data-track/week-11/exercise_4/.env.example create mode 100644 data-track/week-11/exercise_4/README.md create mode 100644 data-track/week-11/exercise_4/app.py rename data-track/{week-12/exercise_3 => week-11/exercise_4}/requirements.txt (51%) create mode 100644 data-track/week-11/exercise_4/solutions/app.py delete mode 100644 data-track/week-12/README.md delete mode 100644 data-track/week-12/exercise_3/.env.example delete mode 100644 data-track/week-12/exercise_3/README.md delete mode 100644 data-track/week-12/exercise_3/app.py delete mode 100644 data-track/week-12/exercise_3/solutions/app.py delete mode 100644 data-track/week-12/exercise_4/README.md delete mode 100644 data-track/week-12/exercise_4/app.py delete mode 100644 data-track/week-12/exercise_4/solutions/app.py diff --git a/data-track/week-11/README.md b/data-track/week-11/README.md new file mode 100644 index 0000000..f7d45f0 --- /dev/null +++ b/data-track/week-11/README.md @@ -0,0 +1,32 @@ +# Week 11: Dashboarding with Metabase and Streamlit + +Practice exercises for [Week 11](https://www.notion.so/hackyourfuture/Data-track) of the Data Track. + + +Both tools read the **dbt marts you built in Week 10** (`fct_trips`) straight from Azure Postgres — +no orchestration required. Metabase gives you point-and-click Questions and dashboards; Streamlit +lets you build a light metrics app in a few lines of Python. + +## Exercises + +| Exercise | Topic | +|---|---| +| [Exercise 1](exercise_1/) | Write a Metabase SQL Question | +| [Exercise 2](exercise_2/) | Build a Metabase dashboard with a date filter | +| [Exercise 3](exercise_3/) | Streamlit: KPI metrics from `fct_trips` | +| [Exercise 4](exercise_4/) | Streamlit: daily trend + data-freshness panel | + +## Prerequisites + +- Logged in to HYF Metabase (URL in the Week 11 chapter) +- Your Week 10 `fct_trips` table is populated in `dev_` on Azure Postgres +- Your Azure Postgres connection string (`PG_URL`) for the Streamlit exercises + +## Running the Streamlit exercises + +```bash +cd exercise_3 # or exercise_4 +pip install -r requirements.txt +cp .env.example .env # fill in PG_URL and PG_SCHEMA +streamlit run app.py +``` diff --git a/data-track/week-12/exercise_1/README.md b/data-track/week-11/exercise_1/README.md similarity index 100% rename from data-track/week-12/exercise_1/README.md rename to data-track/week-11/exercise_1/README.md diff --git a/data-track/week-12/exercise_1/solutions/README.md b/data-track/week-11/exercise_1/solutions/README.md similarity index 100% rename from data-track/week-12/exercise_1/solutions/README.md rename to data-track/week-11/exercise_1/solutions/README.md diff --git a/data-track/week-12/exercise_2/README.md b/data-track/week-11/exercise_2/README.md similarity index 100% rename from data-track/week-12/exercise_2/README.md rename to data-track/week-11/exercise_2/README.md diff --git a/data-track/week-12/exercise_2/solutions/README.md b/data-track/week-11/exercise_2/solutions/README.md similarity index 100% rename from data-track/week-12/exercise_2/solutions/README.md rename to data-track/week-11/exercise_2/solutions/README.md diff --git a/data-track/week-12/exercise_4/.env.example b/data-track/week-11/exercise_3/.env.example similarity index 64% rename from data-track/week-12/exercise_4/.env.example rename to data-track/week-11/exercise_3/.env.example index b92196f..58a7bdd 100644 --- a/data-track/week-12/exercise_4/.env.example +++ b/data-track/week-11/exercise_3/.env.example @@ -1,7 +1,3 @@ -AIRFLOW_URL=https://your-airflow.example.com -AIRFLOW_USER=admin -AIRFLOW_PASS=your-password - # PostgreSQL connection string PG_URL=postgresql://pipeline_user:your-pg-password@your-pg-host:5432/team1 # Your dev schema name (e.g. dev_jana) diff --git a/data-track/week-11/exercise_3/README.md b/data-track/week-11/exercise_3/README.md new file mode 100644 index 0000000..6ed1da8 --- /dev/null +++ b/data-track/week-11/exercise_3/README.md @@ -0,0 +1,54 @@ +# Exercise 3: Streamlit KPI Metrics from `fct_trips` + +Build a Streamlit app that reads your `fct_trips` dbt mart from Azure Postgres and shows three +headline KPI tiles. No orchestration involved — the app queries the table directly. + +## Prerequisites + +- Your Week 10 `fct_trips` table populated in `dev_` on Azure Postgres +- Your Postgres connection string (`PG_URL`) and schema name (`PG_SCHEMA`) from your teacher +- Python 3.11+ and `pip` + +## Setup + +```bash +pip install -r requirements.txt +cp .env.example .env +# Fill in PG_URL and PG_SCHEMA in .env +streamlit run app.py +``` + +## Task + +Open `app.py`. Follow the `TODO` comment to: + +1. Implement `get_trip_metrics(pg_url, schema)` using `psycopg2`. +2. Query `COUNT(*)`, `AVG(fare_amount)`, and `SUM(fare_amount)` from `{schema}.fct_trips`. +3. Return a dict `{"trip_count", "avg_fare", "total_fare"}`; the app renders it as three + `st.metric` tiles (Total trips, Avg fare, Total fare revenue). + +The app should look like this when working: + +```text +┌───────────────┬───────────────┬─────────────────────┐ +│ Total trips │ Avg fare │ Total fare revenue │ +│ 57,000 │ $13.42 │ $765,000 │ +└───────────────┴───────────────┴─────────────────────┘ +``` + +## Success criteria + +- Running `streamlit run app.py` opens a page in the browser. +- The three tiles show non-zero values from your real `fct_trips` data. +- The function uses `@st.cache_data(ttl=60)`. +- No credentials are hardcoded in `app.py` — they come from environment variables only. + +## Stretch + +Add a `st.selectbox` in the sidebar to filter the KPIs by `pickup_borough`. The tiles should +recompute when you change the selection (pass the borough into the query with a parameterised +`WHERE pickup_borough = %s`). + +## Solution + +See `solutions/app.py` for the reference implementation. diff --git a/data-track/week-11/exercise_3/app.py b/data-track/week-11/exercise_3/app.py new file mode 100644 index 0000000..395d576 --- /dev/null +++ b/data-track/week-11/exercise_3/app.py @@ -0,0 +1,42 @@ +""" +Exercise 3: Streamlit KPI metrics panel from fct_trips. +Fill in the TODO to make the tiles show real data. +""" + +import os + +import psycopg2 +import streamlit as st +from dotenv import load_dotenv + +load_dotenv() + +PG_URL = os.environ["PG_URL"] +PG_SCHEMA = os.environ.get("PG_SCHEMA", "dev_yourname") + + +@st.cache_data(ttl=60) +def get_trip_metrics(pg_url: str, schema: str) -> dict: + """Return headline KPIs for fct_trips. + + TODO: implement this function. + Use psycopg2.connect(pg_url) to connect. + Run: SELECT COUNT(*), AVG(fare_amount), SUM(fare_amount) FROM {schema}.fct_trips + Return a dict: {"trip_count": int, "avg_fare": float, "total_fare": float} + """ + raise NotImplementedError("TODO: implement get_trip_metrics") + + +st.title("NYC Taxi Metrics") +st.subheader("Headline KPIs") + +try: + metrics = get_trip_metrics(PG_URL, PG_SCHEMA) + col1, col2, col3 = st.columns(3) + col1.metric("Total trips", f"{metrics['trip_count']:,}") + col2.metric("Avg fare", f"${metrics['avg_fare']:.2f}") + col3.metric("Total fare revenue", f"${metrics['total_fare']:,.0f}") +except NotImplementedError: + st.warning("Implement `get_trip_metrics` to see live data.") +except Exception as exc: + st.error(f"Postgres error: {exc}") diff --git a/data-track/week-12/exercise_4/requirements.txt b/data-track/week-11/exercise_3/requirements.txt similarity index 78% rename from data-track/week-12/exercise_4/requirements.txt rename to data-track/week-11/exercise_3/requirements.txt index 2cf1fd3..8f8459c 100644 --- a/data-track/week-12/exercise_4/requirements.txt +++ b/data-track/week-11/exercise_3/requirements.txt @@ -1,4 +1,3 @@ streamlit>=1.35.0 -requests>=2.31.0 psycopg2-binary>=2.9.9 python-dotenv>=1.0.0 diff --git a/data-track/week-11/exercise_3/solutions/app.py b/data-track/week-11/exercise_3/solutions/app.py new file mode 100644 index 0000000..fd511da --- /dev/null +++ b/data-track/week-11/exercise_3/solutions/app.py @@ -0,0 +1,36 @@ +"""Exercise 3 solution.""" + +import os + +import psycopg2 +import streamlit as st +from dotenv import load_dotenv + +load_dotenv() + +PG_URL = os.environ["PG_URL"] +PG_SCHEMA = os.environ.get("PG_SCHEMA", "dev_yourname") + + +@st.cache_data(ttl=60) +def get_trip_metrics(pg_url: str, schema: str) -> dict: + with psycopg2.connect(pg_url) as conn, conn.cursor() as cur: + cur.execute( + f"SELECT COUNT(*), AVG(fare_amount), SUM(fare_amount) FROM {schema}.fct_trips" + ) + trip_count, avg_fare, total_fare = cur.fetchone() + return { + "trip_count": trip_count or 0, + "avg_fare": float(avg_fare or 0), + "total_fare": float(total_fare or 0), + } + + +st.title("NYC Taxi Metrics") +st.subheader("Headline KPIs") + +metrics = get_trip_metrics(PG_URL, PG_SCHEMA) +col1, col2, col3 = st.columns(3) +col1.metric("Total trips", f"{metrics['trip_count']:,}") +col2.metric("Avg fare", f"${metrics['avg_fare']:.2f}") +col3.metric("Total fare revenue", f"${metrics['total_fare']:,.0f}") diff --git a/data-track/week-11/exercise_4/.env.example b/data-track/week-11/exercise_4/.env.example new file mode 100644 index 0000000..58a7bdd --- /dev/null +++ b/data-track/week-11/exercise_4/.env.example @@ -0,0 +1,4 @@ +# PostgreSQL connection string +PG_URL=postgresql://pipeline_user:your-pg-password@your-pg-host:5432/team1 +# Your dev schema name (e.g. dev_jana) +PG_SCHEMA=dev_yourname diff --git a/data-track/week-11/exercise_4/README.md b/data-track/week-11/exercise_4/README.md new file mode 100644 index 0000000..814e38e --- /dev/null +++ b/data-track/week-11/exercise_4/README.md @@ -0,0 +1,47 @@ +# Exercise 4: Streamlit Daily Trend + Data-Freshness Panel + +Extend the metrics app from Exercise 3 with a daily trip-volume line chart. The data-freshness +panel (row count + last pickup) is already provided so you can focus on the chart query. + +## Prerequisites + +- Exercise 3 working (KPI tiles showing real data) +- Azure Postgres connection string (`PG_URL`) and schema name (`PG_SCHEMA`) from your teacher + +## Setup + +```bash +pip install -r requirements.txt +cp .env.example .env +# Fill in PG_URL and PG_SCHEMA +streamlit run app.py +``` + +## Task + +Open `app.py`. Follow the `TODO` comment to: + +1. Implement `get_daily_trips(pg_url, schema)` using `psycopg2`. +2. Query trips per day: + `SELECT date_trunc('day', pickup_datetime) AS day, COUNT(*) AS trips + FROM {schema}.fct_trips GROUP BY 1 ORDER BY 1`. +3. Return a `pandas.DataFrame` indexed by `day` with a `trips` column; the app renders it with + `st.line_chart`. + +The freshness panel (`get_fct_trips_freshness`) is already implemented — leave it as is. + +## Success criteria + +- The line chart shows trips per day for your `fct_trips` data. +- Both freshness `st.metric` widgets show non-zero / non-null values. +- Both query functions use `@st.cache_data(ttl=60)`. +- Credentials come from environment variables only. + +## Stretch + +Calculate how many hours ago the last pickup was and display it as the `delta` parameter of the +"Last pickup" `st.metric`. For example: "Last pickup: 2024-01-15 03:00 UTC, 14 h ago". + +## Solution + +See `solutions/app.py`. diff --git a/data-track/week-11/exercise_4/app.py b/data-track/week-11/exercise_4/app.py new file mode 100644 index 0000000..3d9eec3 --- /dev/null +++ b/data-track/week-11/exercise_4/app.py @@ -0,0 +1,65 @@ +""" +Exercise 4: add a daily-trend chart to the metrics app. +""" + +import os + +import pandas as pd +import psycopg2 +import streamlit as st +from dotenv import load_dotenv + +load_dotenv() + +PG_URL = os.environ["PG_URL"] +PG_SCHEMA = os.environ.get("PG_SCHEMA", "dev_yourname") + + +@st.cache_data(ttl=60) +def get_daily_trips(pg_url: str, schema: str) -> pd.DataFrame: + """Return daily trip counts for fct_trips. + + TODO: implement this function. + Use psycopg2.connect(pg_url) to connect. + Run: SELECT date_trunc('day', pickup_datetime) AS day, COUNT(*) AS trips + FROM {schema}.fct_trips + GROUP BY 1 + ORDER BY 1 + Return a DataFrame with a "day" index and a "trips" column + (e.g. pd.DataFrame(rows, columns=["day", "trips"]).set_index("day")). + """ + raise NotImplementedError("TODO: implement get_daily_trips") + + +@st.cache_data(ttl=60) +def get_fct_trips_freshness(pg_url: str, schema: str) -> dict: + with psycopg2.connect(pg_url) as conn, conn.cursor() as cur: + cur.execute( + f"SELECT COUNT(*), MAX(pickup_datetime) FROM {schema}.fct_trips" + ) + row_count, last_pickup = cur.fetchone() + return {"row_count": row_count or 0, "last_pickup": last_pickup} + + +st.title("NYC Taxi Metrics") + +# Panel 1: daily trip volume +st.subheader("Daily trip volume") +try: + daily = get_daily_trips(PG_URL, PG_SCHEMA) + st.line_chart(daily) +except NotImplementedError: + st.warning("Implement `get_daily_trips` to see the trend chart.") +except Exception as exc: + st.error(f"Postgres error: {exc}") + +# Panel 2: data freshness +st.subheader("Data freshness") +try: + freshness = get_fct_trips_freshness(PG_URL, PG_SCHEMA) + col1, col2 = st.columns(2) + col1.metric("Row count", f"{freshness['row_count']:,}") + last = freshness["last_pickup"] + col2.metric("Last pickup", str(last) if last else "None") +except Exception as exc: + st.error(f"Postgres error: {exc}") diff --git a/data-track/week-12/exercise_3/requirements.txt b/data-track/week-11/exercise_4/requirements.txt similarity index 51% rename from data-track/week-12/exercise_3/requirements.txt rename to data-track/week-11/exercise_4/requirements.txt index add1236..b530158 100644 --- a/data-track/week-12/exercise_3/requirements.txt +++ b/data-track/week-11/exercise_4/requirements.txt @@ -1,3 +1,4 @@ streamlit>=1.35.0 -requests>=2.31.0 +psycopg2-binary>=2.9.9 +pandas>=2.2.0 python-dotenv>=1.0.0 diff --git a/data-track/week-11/exercise_4/solutions/app.py b/data-track/week-11/exercise_4/solutions/app.py new file mode 100644 index 0000000..5d21451 --- /dev/null +++ b/data-track/week-11/exercise_4/solutions/app.py @@ -0,0 +1,48 @@ +"""Exercise 4 solution.""" + +import os + +import pandas as pd +import psycopg2 +import streamlit as st +from dotenv import load_dotenv + +load_dotenv() + +PG_URL = os.environ["PG_URL"] +PG_SCHEMA = os.environ.get("PG_SCHEMA", "dev_yourname") + + +@st.cache_data(ttl=60) +def get_daily_trips(pg_url: str, schema: str) -> pd.DataFrame: + with psycopg2.connect(pg_url) as conn, conn.cursor() as cur: + cur.execute( + f"SELECT date_trunc('day', pickup_datetime) AS day, COUNT(*) AS trips " + f"FROM {schema}.fct_trips GROUP BY 1 ORDER BY 1" + ) + rows = cur.fetchall() + return pd.DataFrame(rows, columns=["day", "trips"]).set_index("day") + + +@st.cache_data(ttl=60) +def get_fct_trips_freshness(pg_url: str, schema: str) -> dict: + with psycopg2.connect(pg_url) as conn, conn.cursor() as cur: + cur.execute( + f"SELECT COUNT(*), MAX(pickup_datetime) FROM {schema}.fct_trips" + ) + row_count, last_pickup = cur.fetchone() + return {"row_count": row_count or 0, "last_pickup": last_pickup} + + +st.title("NYC Taxi Metrics") + +st.subheader("Daily trip volume") +daily = get_daily_trips(PG_URL, PG_SCHEMA) +st.line_chart(daily) + +st.subheader("Data freshness") +freshness = get_fct_trips_freshness(PG_URL, PG_SCHEMA) +col1, col2 = st.columns(2) +col1.metric("Row count", f"{freshness['row_count']:,}") +last = freshness["last_pickup"] +col2.metric("Last pickup", str(last) if last else "None") diff --git a/data-track/week-12/README.md b/data-track/week-12/README.md deleted file mode 100644 index 7269e14..0000000 --- a/data-track/week-12/README.md +++ /dev/null @@ -1,27 +0,0 @@ -# Week 12: Dashboarding with Metabase and Streamlit - -Practice exercises for [Week 12](https://www.notion.so/hackyourfuture/Week-12-Dashboarding-with-Metabase-and-Streamlit) of the Data Track. - -## Exercises - -| Exercise | Topic | -|---|---| -| [Exercise 1](exercise_1/) | Write a Metabase SQL Question | -| [Exercise 2](exercise_2/) | Build a Metabase dashboard with a date filter | -| [Exercise 3](exercise_3/) | Streamlit: last DAG run status panel | -| [Exercise 4](exercise_4/) | Streamlit: data-freshness panel from Postgres | - -## Prerequisites - -- Logged in to HYF Metabase (URL in Week 12, Ch2) -- Your Week 10 `fct_trips` table is populated in `dev_` on Azure Postgres -- Your Week 11 Airflow `ingest_taxi_month` DAG has at least one run - -## Running the Streamlit exercises - -```bash -cd exercise_3 # or exercise_4 -pip install -r requirements.txt -cp .env.example .env # fill in your credentials -streamlit run app.py -``` diff --git a/data-track/week-12/exercise_3/.env.example b/data-track/week-12/exercise_3/.env.example deleted file mode 100644 index 278f575..0000000 --- a/data-track/week-12/exercise_3/.env.example +++ /dev/null @@ -1,3 +0,0 @@ -AIRFLOW_URL=https://your-airflow.example.com -AIRFLOW_USER=admin -AIRFLOW_PASS=your-password diff --git a/data-track/week-12/exercise_3/README.md b/data-track/week-12/exercise_3/README.md deleted file mode 100644 index 18ae3bd..0000000 --- a/data-track/week-12/exercise_3/README.md +++ /dev/null @@ -1,47 +0,0 @@ -# Exercise 3: Streamlit Last DAG Run Status Panel - -Build a Streamlit app with one panel that calls the Airflow REST API and shows the last run state. - -## Prerequisites - -- Airflow URL, username, and password (from your teacher) -- Python 3.11+ and `pip` - -## Setup - -```bash -pip install -r requirements.txt -cp .env.example .env -# Fill in AIRFLOW_URL, AIRFLOW_USER, AIRFLOW_PASS in .env -streamlit run app.py -``` - -## Task - -Open `app.py`. Follow the `TODO` comments to: - -1. Implement `get_dag_runs(dag_id, limit)`: Airflow 3 uses token auth, so first `POST` to `/auth/token` for a JWT, then call `/api/v2/.../dagRuns` with an `Authorization: Bearer ` header. -2. Call it with `dag_id = "ingest_taxi_month"` and `limit = 1`. -3. Display the result as `st.success` (success), `st.error` (failed), or `st.warning` (other state). - -The app should look like this when working: - -```text -┌─────────────────────────────────────────┐ -│ Last run: success - started 2024-01-15 │ -└─────────────────────────────────────────┘ -``` - -## Success criteria - -- Running `streamlit run app.py` opens a page in the browser. -- The panel shows the real last run state of your `ingest_taxi_month` DAG. -- No credentials are hardcoded in `app.py`. - -## Stretch - -Add a `st.selectbox` in the sidebar to switch between two DAG IDs. The panel should update when you change the selection. - -## Solution - -See `solutions/app.py` for the reference implementation. diff --git a/data-track/week-12/exercise_3/app.py b/data-track/week-12/exercise_3/app.py deleted file mode 100644 index fbe23f0..0000000 --- a/data-track/week-12/exercise_3/app.py +++ /dev/null @@ -1,55 +0,0 @@ -""" -Exercise 3: Streamlit last DAG run status panel. -Fill in the TODOs to make the panel show real data. -""" - -import os - -import requests -import streamlit as st -from dotenv import load_dotenv - -load_dotenv() - -AIRFLOW_URL = os.environ["AIRFLOW_URL"] -AIRFLOW_USER = os.environ["AIRFLOW_USER"] -AIRFLOW_PASS = os.environ["AIRFLOW_PASS"] - - -@st.cache_data(ttl=60) -def get_dag_runs(dag_id: str, limit: int = 10) -> list: - """Return recent DAG runs from the Airflow REST API. - - TODO: implement this function. - Endpoint: GET {AIRFLOW_URL}/api/v2/dags/{dag_id}/dagRuns - Params: limit=limit, order_by="-logical_date" - Auth: Airflow 3 uses a token, not Basic Auth. First POST to - {AIRFLOW_URL}/auth/token with {"username":..., "password":...} to get - an access_token, then send headers={"Authorization": f"Bearer {token}"}. - Return: response.json()["dag_runs"] - """ - raise NotImplementedError("TODO: implement get_dag_runs") - - -st.title("Pipeline Health") -st.subheader("Last DAG run") - -dag_id = "ingest_taxi_month" - -try: - runs = get_dag_runs(dag_id, limit=1) - if runs: - last = runs[0] - state = last["state"] - if state == "success": - st.success(f"Last run: **{state}**, started {last['start_date']}") - elif state == "failed": - st.error(f"Last run: **{state}**, check Airflow logs") - else: - st.warning(f"Last run: **{state}**") - else: - st.info("No runs found.") -except NotImplementedError: - st.warning("Implement `get_dag_runs` to see live data.") -except Exception as exc: - st.error(f"Error: {exc}") diff --git a/data-track/week-12/exercise_3/solutions/app.py b/data-track/week-12/exercise_3/solutions/app.py deleted file mode 100644 index b44ccdf..0000000 --- a/data-track/week-12/exercise_3/solutions/app.py +++ /dev/null @@ -1,55 +0,0 @@ -"""Exercise 3 solution.""" - -import os - -import requests -import streamlit as st -from dotenv import load_dotenv - -load_dotenv() - -AIRFLOW_URL = os.environ["AIRFLOW_URL"] -AIRFLOW_USER = os.environ["AIRFLOW_USER"] -AIRFLOW_PASS = os.environ["AIRFLOW_PASS"] - - -@st.cache_data(ttl=3000) # Airflow 3 uses token auth; the token is valid for 24h -def get_airflow_token() -> str: - r = requests.post( - f"{AIRFLOW_URL}/auth/token", - json={"username": AIRFLOW_USER, "password": AIRFLOW_PASS}, - ) - r.raise_for_status() - return r.json()["access_token"] - - -@st.cache_data(ttl=60) -def get_dag_runs(dag_id: str, limit: int = 10) -> list: - url = f"{AIRFLOW_URL}/api/v2/dags/{dag_id}/dagRuns" - r = requests.get( - url, - params={"limit": limit, "order_by": "-logical_date"}, - headers={"Authorization": f"Bearer {get_airflow_token()}"}, - timeout=10, - ) - r.raise_for_status() - return r.json()["dag_runs"] - - -st.title("Pipeline Health") -st.subheader("Last DAG run") - -dag_id = "ingest_taxi_month" - -runs = get_dag_runs(dag_id, limit=1) -if runs: - last = runs[0] - state = last["state"] - if state == "success": - st.success(f"Last run: **{state}**, started {last['start_date']}") - elif state == "failed": - st.error(f"Last run: **{state}**, check Airflow logs") - else: - st.warning(f"Last run: **{state}**") -else: - st.info("No runs found.") diff --git a/data-track/week-12/exercise_4/README.md b/data-track/week-12/exercise_4/README.md deleted file mode 100644 index 5783d3e..0000000 --- a/data-track/week-12/exercise_4/README.md +++ /dev/null @@ -1,39 +0,0 @@ -# Exercise 4: Streamlit Data-Freshness Panel - -Extend the app from Exercise 3 with a second panel that reads `fct_trips` row count and last-updated timestamp from Azure Postgres. - -## Prerequisites - -- Exercise 3 working (last run panel showing real data) -- Azure Postgres connection string (`PG_URL`) from your teacher - -## Setup - -```bash -pip install -r requirements.txt -cp .env.example .env -# Add PG_URL to .env (in addition to the Airflow env vars) -streamlit run app.py -``` - -## Task - -Open `app.py`. Follow the `TODO` comments to: - -1. Implement `get_fct_trips_freshness(pg_url)` using `psycopg2`. -2. Query `MAX(pickup_datetime)` and `COUNT(*)` from `dev_.fct_trips`. -3. Display as two `st.metric` widgets: "Row count" and "Last pickup". - -## Success criteria - -- Both `st.metric` widgets show non-zero / non-null values from real data. -- The function uses `@st.cache_data(ttl=60)`. -- Credentials come from environment variables only. - -## Stretch - -Calculate how many hours ago the last pickup was and display it as the `delta` parameter of `st.metric`. For example: "Last pickup: 2024-01-15 03:00 UTC, 14 h ago". - -## Solution - -See `solutions/app.py`. diff --git a/data-track/week-12/exercise_4/app.py b/data-track/week-12/exercise_4/app.py deleted file mode 100644 index e814a65..0000000 --- a/data-track/week-12/exercise_4/app.py +++ /dev/null @@ -1,87 +0,0 @@ -""" -Exercise 4: add a data-freshness panel to the pipeline health app. -""" - -import os -from datetime import datetime, timezone - -import psycopg2 -import requests -import streamlit as st -from dotenv import load_dotenv - -load_dotenv() - -AIRFLOW_URL = os.environ["AIRFLOW_URL"] -AIRFLOW_USER = os.environ["AIRFLOW_USER"] -AIRFLOW_PASS = os.environ["AIRFLOW_PASS"] - - -@st.cache_data(ttl=3000) # Airflow 3 uses token auth; the token is valid for 24h -def get_airflow_token() -> str: - r = requests.post( - f"{AIRFLOW_URL}/auth/token", - json={"username": AIRFLOW_USER, "password": AIRFLOW_PASS}, - ) - r.raise_for_status() - return r.json()["access_token"] -PG_URL = os.environ["PG_URL"] -PG_SCHEMA = os.environ.get("PG_SCHEMA", "dev_yourname") - - -@st.cache_data(ttl=60) -def get_dag_runs(dag_id: str, limit: int = 10) -> list: - url = f"{AIRFLOW_URL}/api/v2/dags/{dag_id}/dagRuns" - r = requests.get( - url, - params={"limit": limit, "order_by": "-logical_date"}, - headers={"Authorization": f"Bearer {get_airflow_token()}"}, - timeout=10, - ) - r.raise_for_status() - return r.json()["dag_runs"] - - -@st.cache_data(ttl=60) -def get_fct_trips_freshness(pg_url: str, schema: str) -> dict: - """Return row count and last pickup_datetime for fct_trips. - - TODO: implement this function. - Use psycopg2.connect(pg_url) to connect. - Run: SELECT COUNT(*), MAX(pickup_datetime) FROM {schema}.fct_trips - Return a dict: {"row_count": int, "last_pickup": datetime or None} - """ - raise NotImplementedError("TODO: implement get_fct_trips_freshness") - - -st.title("Pipeline Health") - -# Panel 1: last run -st.subheader("Last DAG run") -dag_id = "ingest_taxi_month" -try: - runs = get_dag_runs(dag_id, limit=1) - if runs: - last = runs[0] - state = last["state"] - if state == "success": - st.success(f"Last run: **{state}**, started {last['start_date']}") - elif state == "failed": - st.error(f"Last run: **{state}**, check Airflow logs") - else: - st.warning(f"Last run: **{state}**") -except Exception as exc: - st.error(f"Airflow error: {exc}") - -# Panel 2: data freshness -st.subheader("Data freshness") -try: - freshness = get_fct_trips_freshness(PG_URL, PG_SCHEMA) - col1, col2 = st.columns(2) - col1.metric("Row count", f"{freshness['row_count']:,}") - last = freshness["last_pickup"] - col2.metric("Last pickup", str(last) if last else "None") -except NotImplementedError: - st.warning("Implement `get_fct_trips_freshness` to see freshness data.") -except Exception as exc: - st.error(f"Postgres error: {exc}") diff --git a/data-track/week-12/exercise_4/solutions/app.py b/data-track/week-12/exercise_4/solutions/app.py deleted file mode 100644 index 16a2c05..0000000 --- a/data-track/week-12/exercise_4/solutions/app.py +++ /dev/null @@ -1,72 +0,0 @@ -"""Exercise 4 solution.""" - -import os - -import psycopg2 -import requests -import streamlit as st -from dotenv import load_dotenv - -load_dotenv() - -AIRFLOW_URL = os.environ["AIRFLOW_URL"] -AIRFLOW_USER = os.environ["AIRFLOW_USER"] -AIRFLOW_PASS = os.environ["AIRFLOW_PASS"] - - -@st.cache_data(ttl=3000) # Airflow 3 uses token auth; the token is valid for 24h -def get_airflow_token() -> str: - r = requests.post( - f"{AIRFLOW_URL}/auth/token", - json={"username": AIRFLOW_USER, "password": AIRFLOW_PASS}, - ) - r.raise_for_status() - return r.json()["access_token"] -PG_URL = os.environ["PG_URL"] -PG_SCHEMA = os.environ.get("PG_SCHEMA", "dev_yourname") - - -@st.cache_data(ttl=60) -def get_dag_runs(dag_id: str, limit: int = 10) -> list: - url = f"{AIRFLOW_URL}/api/v2/dags/{dag_id}/dagRuns" - r = requests.get( - url, - params={"limit": limit, "order_by": "-logical_date"}, - headers={"Authorization": f"Bearer {get_airflow_token()}"}, - timeout=10, - ) - r.raise_for_status() - return r.json()["dag_runs"] - - -@st.cache_data(ttl=60) -def get_fct_trips_freshness(pg_url: str, schema: str) -> dict: - with psycopg2.connect(pg_url) as conn, conn.cursor() as cur: - cur.execute( - f"SELECT COUNT(*), MAX(pickup_datetime) FROM {schema}.fct_trips" - ) - row_count, last_pickup = cur.fetchone() - return {"row_count": row_count or 0, "last_pickup": last_pickup} - - -st.title("Pipeline Health") - -st.subheader("Last DAG run") -dag_id = "ingest_taxi_month" -runs = get_dag_runs(dag_id, limit=1) -if runs: - last = runs[0] - state = last["state"] - if state == "success": - st.success(f"Last run: **{state}**, started {last['start_date']}") - elif state == "failed": - st.error(f"Last run: **{state}**, check Airflow logs") - else: - st.warning(f"Last run: **{state}**") - -st.subheader("Data freshness") -freshness = get_fct_trips_freshness(PG_URL, PG_SCHEMA) -col1, col2 = st.columns(2) -col1.metric("Row count", f"{freshness['row_count']:,}") -last = freshness["last_pickup"] -col2.metric("Last pickup", str(last) if last else "None") From e884d410148c61660cdfd11813f3d8d906a26dfb Mon Sep 17 00:00:00 2001 From: Lasse Benninga Date: Tue, 7 Jul 2026 15:18:19 +0200 Subject: [PATCH 20/20] chore(data-track/week-11): move dashboarding exercises to reference repos MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit From week 11+ the Data Track uses external reference repos with exercise branches instead of one-off exercise folders in Learning-Resources (mirrors the Week 10 dbt reference repo). Replace the week-11 exercise_N folders with a pointer README linking to: - lassebenni/nyc-taxi-streamlit-reference (KPI tiles, daily-trend chart) - lassebenni/nyc-taxi-metabase-reference (SQL Question, dashboard + filter) Both read the Week 10 fct_trips dbt mart from Postgres — no Airflow. Co-Authored-By: Claude Opus 4.8 (1M context) --- data-track/week-11/README.md | 58 ++++++++++++----- data-track/week-11/exercise_1/README.md | 41 ------------ .../week-11/exercise_1/solutions/README.md | 24 ------- data-track/week-11/exercise_2/README.md | 41 ------------ .../week-11/exercise_2/solutions/README.md | 15 ----- data-track/week-11/exercise_3/.env.example | 4 -- data-track/week-11/exercise_3/README.md | 54 --------------- data-track/week-11/exercise_3/app.py | 42 ------------ .../week-11/exercise_3/requirements.txt | 3 - .../week-11/exercise_3/solutions/app.py | 36 ---------- data-track/week-11/exercise_4/.env.example | 4 -- data-track/week-11/exercise_4/README.md | 47 -------------- data-track/week-11/exercise_4/app.py | 65 ------------------- .../week-11/exercise_4/requirements.txt | 4 -- .../week-11/exercise_4/solutions/app.py | 48 -------------- 15 files changed, 40 insertions(+), 446 deletions(-) delete mode 100644 data-track/week-11/exercise_1/README.md delete mode 100644 data-track/week-11/exercise_1/solutions/README.md delete mode 100644 data-track/week-11/exercise_2/README.md delete mode 100644 data-track/week-11/exercise_2/solutions/README.md delete mode 100644 data-track/week-11/exercise_3/.env.example delete mode 100644 data-track/week-11/exercise_3/README.md delete mode 100644 data-track/week-11/exercise_3/app.py delete mode 100644 data-track/week-11/exercise_3/requirements.txt delete mode 100644 data-track/week-11/exercise_3/solutions/app.py delete mode 100644 data-track/week-11/exercise_4/.env.example delete mode 100644 data-track/week-11/exercise_4/README.md delete mode 100644 data-track/week-11/exercise_4/app.py delete mode 100644 data-track/week-11/exercise_4/requirements.txt delete mode 100644 data-track/week-11/exercise_4/solutions/app.py diff --git a/data-track/week-11/README.md b/data-track/week-11/README.md index f7d45f0..e9b791c 100644 --- a/data-track/week-11/README.md +++ b/data-track/week-11/README.md @@ -3,30 +3,52 @@ Practice exercises for [Week 11](https://www.notion.so/hackyourfuture/Data-track) of the Data Track. -Both tools read the **dbt marts you built in Week 10** (`fct_trips`) straight from Azure Postgres — -no orchestration required. Metabase gives you point-and-click Questions and dashboards; Streamlit -lets you build a light metrics app in a few lines of Python. +The Week 11 practice no longer lives here as loose exercise folders. Like the Week 10 dbt drills, it +now lives as **branches in dedicated reference repos**, each starting from a complete, runnable +project so you can actually run the app / test the SQL instead of grafting snippets into your own +setup. -## Exercises +Both tools read the **dbt mart you built in Week 10** (`fct_trips`) straight from Azure Postgres — +no orchestration required. Clone a repo once, `git switch` to an exercise branch, follow its +`EXERCISE.md`, then compare against the matching `-solution` branch. -| Exercise | Topic | -|---|---| -| [Exercise 1](exercise_1/) | Write a Metabase SQL Question | -| [Exercise 2](exercise_2/) | Build a Metabase dashboard with a date filter | -| [Exercise 3](exercise_3/) | Streamlit: KPI metrics from `fct_trips` | -| [Exercise 4](exercise_4/) | Streamlit: daily trend + data-freshness panel | +## Streamlit — [`nyc-taxi-streamlit-reference`](https://github.com/lassebenni/nyc-taxi-streamlit-reference) -## Prerequisites - -- Logged in to HYF Metabase (URL in the Week 11 chapter) -- Your Week 10 `fct_trips` table is populated in `dev_` on Azure Postgres -- Your Azure Postgres connection string (`PG_URL`) for the Streamlit exercises +A runnable metrics app (KPI tiles, daily-trend chart, freshness panel) that queries `fct_trips`. -## Running the Streamlit exercises +| Exercise | Start branch | Solution branch | +| --- | --- | --- | +| KPI tiles from `fct_trips` | [`exercise-kpi-metrics`](https://github.com/lassebenni/nyc-taxi-streamlit-reference/tree/exercise-kpi-metrics) | [`exercise-kpi-metrics-solution`](https://github.com/lassebenni/nyc-taxi-streamlit-reference/tree/exercise-kpi-metrics-solution) | +| Daily trip-volume chart | [`exercise-daily-trend`](https://github.com/lassebenni/nyc-taxi-streamlit-reference/tree/exercise-daily-trend) | [`exercise-daily-trend-solution`](https://github.com/lassebenni/nyc-taxi-streamlit-reference/tree/exercise-daily-trend-solution) | ```bash -cd exercise_3 # or exercise_4 +git clone https://github.com/lassebenni/nyc-taxi-streamlit-reference.git +cd nyc-taxi-streamlit-reference +git switch exercise-kpi-metrics +python -m venv .venv && source .venv/bin/activate pip install -r requirements.txt -cp .env.example .env # fill in PG_URL and PG_SCHEMA +cp .env.example .env # set your Week 9/10 PG_URL + PG_SCHEMA streamlit run app.py ``` + +## Metabase — [`nyc-taxi-metabase-reference`](https://github.com/lassebenni/nyc-taxi-metabase-reference) + +Each Metabase Question is driven by a SQL query. Those queries live as runnable `.sql` (your +checkable source of truth); the dashboard assembly stays in the Metabase UI. + +| Exercise | Start branch | Solution branch | +| --- | --- | --- | +| Write a Metabase SQL Question | [`exercise-sql-question`](https://github.com/lassebenni/nyc-taxi-metabase-reference/tree/exercise-sql-question) | [`exercise-sql-question-solution`](https://github.com/lassebenni/nyc-taxi-metabase-reference/tree/exercise-sql-question-solution) | +| Build a dashboard with a date filter | [`exercise-dashboard-filter`](https://github.com/lassebenni/nyc-taxi-metabase-reference/tree/exercise-dashboard-filter) | [`exercise-dashboard-filter-solution`](https://github.com/lassebenni/nyc-taxi-metabase-reference/tree/exercise-dashboard-filter-solution) | + +```bash +git clone https://github.com/lassebenni/nyc-taxi-metabase-reference.git +cd nyc-taxi-metabase-reference +git switch exercise-sql-question # then read EXERCISE.md and log in to Metabase +``` + +## Prerequisites + +- Your Week 10 `fct_trips` table populated in `dev_` on the shared Azure Postgres. +- For Streamlit: your Postgres connection string (`PG_URL`) and schema name (`PG_SCHEMA`). +- For Metabase: logged in to HYF Metabase (URL in the Week 11 chapter). diff --git a/data-track/week-11/exercise_1/README.md b/data-track/week-11/exercise_1/README.md deleted file mode 100644 index 1bc2844..0000000 --- a/data-track/week-11/exercise_1/README.md +++ /dev/null @@ -1,41 +0,0 @@ -# Exercise 1: Write a Metabase SQL Question - -Practice writing a SQL Question in Metabase using your `fct_trips` dbt mart table. - -## Prerequisites - -- Logged in to HYF Metabase -- Your `dev_.fct_trips` table is visible under Browse data - -## Task - -1. Log in to Metabase and go to **New → SQL query**. -2. Select the **Azure Postgres (HYF)** database. -3. Write a query that returns trip count by `pickup_borough`, ordered descending: - - ```sql - SELECT pickup_borough, - COUNT(*) AS trip_count - FROM dev_yourname.fct_trips - GROUP BY pickup_borough - ORDER BY trip_count DESC - ``` - - Replace `dev_yourname` with your actual schema name (e.g. `dev_jana`). - -4. Set the visualisation to **Bar chart**. -5. Save the Question to **Your Personal Collection** with the name "Trip count by borough". - -## Success criteria - -- The Question runs without error. -- The bar chart shows at least 3 boroughs with non-zero counts. -- The Question is saved (you can see it in your Personal Collection). - -## Stretch - -Write a second Question: average `fare_amount` by `pickup_borough`, filtered to exclude trips where `fare_amount = 0`. Save it as "Average fare by borough (non-zero)". - -## Solution - -See `solutions/README.md` for the reference SQL. diff --git a/data-track/week-11/exercise_1/solutions/README.md b/data-track/week-11/exercise_1/solutions/README.md deleted file mode 100644 index da89f6a..0000000 --- a/data-track/week-11/exercise_1/solutions/README.md +++ /dev/null @@ -1,24 +0,0 @@ -# Exercise 1 Solution - -## Minimum SQL - -```sql -SELECT pickup_borough, - COUNT(*) AS trip_count -FROM dev_yourname.fct_trips -GROUP BY pickup_borough -ORDER BY trip_count DESC -``` - -## Stretch SQL (average fare, excluding zero fares) - -```sql -SELECT pickup_borough, - AVG(fare_amount) AS avg_fare -FROM dev_yourname.fct_trips -WHERE fare_amount > 0 -GROUP BY pickup_borough -ORDER BY avg_fare DESC -``` - -Both queries should be saved as SQL Questions in your Personal Collection before proceeding to Exercise 2. diff --git a/data-track/week-11/exercise_2/README.md b/data-track/week-11/exercise_2/README.md deleted file mode 100644 index 110bf15..0000000 --- a/data-track/week-11/exercise_2/README.md +++ /dev/null @@ -1,41 +0,0 @@ -# Exercise 2: Build a Metabase Dashboard with a Date Filter - -Combine the Questions from Exercise 1 into a dashboard, then add a date filter. - -## Prerequisites - -- Exercise 1 completed: "Trip count by borough" Question saved in your Personal Collection -- At least one more Question saved (e.g. "Average fare by borough") - -## Task - -### Part A: Create the dashboard - -1. Go to **New → Dashboard**. -2. Name it "NYC Taxi Analytics: [Your Name]". -3. Click **Add a question** and add both Questions from Exercise 1. -4. Arrange the cards side by side using the drag handles. -5. Click **Done** to exit edit mode. - -### Part B: Add a date filter - -1. Re-enter edit mode (**Edit**). -2. Click **Filters → Date → Date Range**. -3. Open the filter settings for the "Trip count by borough" Question. -4. Map the filter to the `pickup_datetime` column. -5. Repeat for the "Average fare by borough" Question. -6. Click **Done** and test the filter: select a 30-day range and verify both panels update. - -## Success criteria - -- Dashboard contains at least 2 Questions. -- Date-range filter is connected to at least 2 Questions. -- Filtering to a specific month changes the trip counts shown. - -## Stretch - -Add a third Question: daily trip volume over time (line chart). Check that the date filter also connects to it. - -## Solution - -See `solutions/README.md` for step-by-step guidance. diff --git a/data-track/week-11/exercise_2/solutions/README.md b/data-track/week-11/exercise_2/solutions/README.md deleted file mode 100644 index 3c6a38d..0000000 --- a/data-track/week-11/exercise_2/solutions/README.md +++ /dev/null @@ -1,15 +0,0 @@ -# Exercise 2 Solution - -## Key steps - -1. Create dashboard via New → Dashboard. -2. Add Questions via the + button in edit mode. -3. Add filter: Edit → Filters → Date → Date Range. -4. Map filter: click the filter gear icon on each Question card → select the date column (`pickup_datetime`). -5. Done → test. - -## Common issue: filter has no effect - -If the filter does not change the results, the most likely cause is that you forgot to map the filter to the Question's date column. Each Question has its own mapping, adding the filter to the dashboard is not enough. - -Check: enter edit mode → click the filter label → confirm both Question cards show the mapped column. diff --git a/data-track/week-11/exercise_3/.env.example b/data-track/week-11/exercise_3/.env.example deleted file mode 100644 index 58a7bdd..0000000 --- a/data-track/week-11/exercise_3/.env.example +++ /dev/null @@ -1,4 +0,0 @@ -# PostgreSQL connection string -PG_URL=postgresql://pipeline_user:your-pg-password@your-pg-host:5432/team1 -# Your dev schema name (e.g. dev_jana) -PG_SCHEMA=dev_yourname diff --git a/data-track/week-11/exercise_3/README.md b/data-track/week-11/exercise_3/README.md deleted file mode 100644 index 6ed1da8..0000000 --- a/data-track/week-11/exercise_3/README.md +++ /dev/null @@ -1,54 +0,0 @@ -# Exercise 3: Streamlit KPI Metrics from `fct_trips` - -Build a Streamlit app that reads your `fct_trips` dbt mart from Azure Postgres and shows three -headline KPI tiles. No orchestration involved — the app queries the table directly. - -## Prerequisites - -- Your Week 10 `fct_trips` table populated in `dev_` on Azure Postgres -- Your Postgres connection string (`PG_URL`) and schema name (`PG_SCHEMA`) from your teacher -- Python 3.11+ and `pip` - -## Setup - -```bash -pip install -r requirements.txt -cp .env.example .env -# Fill in PG_URL and PG_SCHEMA in .env -streamlit run app.py -``` - -## Task - -Open `app.py`. Follow the `TODO` comment to: - -1. Implement `get_trip_metrics(pg_url, schema)` using `psycopg2`. -2. Query `COUNT(*)`, `AVG(fare_amount)`, and `SUM(fare_amount)` from `{schema}.fct_trips`. -3. Return a dict `{"trip_count", "avg_fare", "total_fare"}`; the app renders it as three - `st.metric` tiles (Total trips, Avg fare, Total fare revenue). - -The app should look like this when working: - -```text -┌───────────────┬───────────────┬─────────────────────┐ -│ Total trips │ Avg fare │ Total fare revenue │ -│ 57,000 │ $13.42 │ $765,000 │ -└───────────────┴───────────────┴─────────────────────┘ -``` - -## Success criteria - -- Running `streamlit run app.py` opens a page in the browser. -- The three tiles show non-zero values from your real `fct_trips` data. -- The function uses `@st.cache_data(ttl=60)`. -- No credentials are hardcoded in `app.py` — they come from environment variables only. - -## Stretch - -Add a `st.selectbox` in the sidebar to filter the KPIs by `pickup_borough`. The tiles should -recompute when you change the selection (pass the borough into the query with a parameterised -`WHERE pickup_borough = %s`). - -## Solution - -See `solutions/app.py` for the reference implementation. diff --git a/data-track/week-11/exercise_3/app.py b/data-track/week-11/exercise_3/app.py deleted file mode 100644 index 395d576..0000000 --- a/data-track/week-11/exercise_3/app.py +++ /dev/null @@ -1,42 +0,0 @@ -""" -Exercise 3: Streamlit KPI metrics panel from fct_trips. -Fill in the TODO to make the tiles show real data. -""" - -import os - -import psycopg2 -import streamlit as st -from dotenv import load_dotenv - -load_dotenv() - -PG_URL = os.environ["PG_URL"] -PG_SCHEMA = os.environ.get("PG_SCHEMA", "dev_yourname") - - -@st.cache_data(ttl=60) -def get_trip_metrics(pg_url: str, schema: str) -> dict: - """Return headline KPIs for fct_trips. - - TODO: implement this function. - Use psycopg2.connect(pg_url) to connect. - Run: SELECT COUNT(*), AVG(fare_amount), SUM(fare_amount) FROM {schema}.fct_trips - Return a dict: {"trip_count": int, "avg_fare": float, "total_fare": float} - """ - raise NotImplementedError("TODO: implement get_trip_metrics") - - -st.title("NYC Taxi Metrics") -st.subheader("Headline KPIs") - -try: - metrics = get_trip_metrics(PG_URL, PG_SCHEMA) - col1, col2, col3 = st.columns(3) - col1.metric("Total trips", f"{metrics['trip_count']:,}") - col2.metric("Avg fare", f"${metrics['avg_fare']:.2f}") - col3.metric("Total fare revenue", f"${metrics['total_fare']:,.0f}") -except NotImplementedError: - st.warning("Implement `get_trip_metrics` to see live data.") -except Exception as exc: - st.error(f"Postgres error: {exc}") diff --git a/data-track/week-11/exercise_3/requirements.txt b/data-track/week-11/exercise_3/requirements.txt deleted file mode 100644 index 8f8459c..0000000 --- a/data-track/week-11/exercise_3/requirements.txt +++ /dev/null @@ -1,3 +0,0 @@ -streamlit>=1.35.0 -psycopg2-binary>=2.9.9 -python-dotenv>=1.0.0 diff --git a/data-track/week-11/exercise_3/solutions/app.py b/data-track/week-11/exercise_3/solutions/app.py deleted file mode 100644 index fd511da..0000000 --- a/data-track/week-11/exercise_3/solutions/app.py +++ /dev/null @@ -1,36 +0,0 @@ -"""Exercise 3 solution.""" - -import os - -import psycopg2 -import streamlit as st -from dotenv import load_dotenv - -load_dotenv() - -PG_URL = os.environ["PG_URL"] -PG_SCHEMA = os.environ.get("PG_SCHEMA", "dev_yourname") - - -@st.cache_data(ttl=60) -def get_trip_metrics(pg_url: str, schema: str) -> dict: - with psycopg2.connect(pg_url) as conn, conn.cursor() as cur: - cur.execute( - f"SELECT COUNT(*), AVG(fare_amount), SUM(fare_amount) FROM {schema}.fct_trips" - ) - trip_count, avg_fare, total_fare = cur.fetchone() - return { - "trip_count": trip_count or 0, - "avg_fare": float(avg_fare or 0), - "total_fare": float(total_fare or 0), - } - - -st.title("NYC Taxi Metrics") -st.subheader("Headline KPIs") - -metrics = get_trip_metrics(PG_URL, PG_SCHEMA) -col1, col2, col3 = st.columns(3) -col1.metric("Total trips", f"{metrics['trip_count']:,}") -col2.metric("Avg fare", f"${metrics['avg_fare']:.2f}") -col3.metric("Total fare revenue", f"${metrics['total_fare']:,.0f}") diff --git a/data-track/week-11/exercise_4/.env.example b/data-track/week-11/exercise_4/.env.example deleted file mode 100644 index 58a7bdd..0000000 --- a/data-track/week-11/exercise_4/.env.example +++ /dev/null @@ -1,4 +0,0 @@ -# PostgreSQL connection string -PG_URL=postgresql://pipeline_user:your-pg-password@your-pg-host:5432/team1 -# Your dev schema name (e.g. dev_jana) -PG_SCHEMA=dev_yourname diff --git a/data-track/week-11/exercise_4/README.md b/data-track/week-11/exercise_4/README.md deleted file mode 100644 index 814e38e..0000000 --- a/data-track/week-11/exercise_4/README.md +++ /dev/null @@ -1,47 +0,0 @@ -# Exercise 4: Streamlit Daily Trend + Data-Freshness Panel - -Extend the metrics app from Exercise 3 with a daily trip-volume line chart. The data-freshness -panel (row count + last pickup) is already provided so you can focus on the chart query. - -## Prerequisites - -- Exercise 3 working (KPI tiles showing real data) -- Azure Postgres connection string (`PG_URL`) and schema name (`PG_SCHEMA`) from your teacher - -## Setup - -```bash -pip install -r requirements.txt -cp .env.example .env -# Fill in PG_URL and PG_SCHEMA -streamlit run app.py -``` - -## Task - -Open `app.py`. Follow the `TODO` comment to: - -1. Implement `get_daily_trips(pg_url, schema)` using `psycopg2`. -2. Query trips per day: - `SELECT date_trunc('day', pickup_datetime) AS day, COUNT(*) AS trips - FROM {schema}.fct_trips GROUP BY 1 ORDER BY 1`. -3. Return a `pandas.DataFrame` indexed by `day` with a `trips` column; the app renders it with - `st.line_chart`. - -The freshness panel (`get_fct_trips_freshness`) is already implemented — leave it as is. - -## Success criteria - -- The line chart shows trips per day for your `fct_trips` data. -- Both freshness `st.metric` widgets show non-zero / non-null values. -- Both query functions use `@st.cache_data(ttl=60)`. -- Credentials come from environment variables only. - -## Stretch - -Calculate how many hours ago the last pickup was and display it as the `delta` parameter of the -"Last pickup" `st.metric`. For example: "Last pickup: 2024-01-15 03:00 UTC, 14 h ago". - -## Solution - -See `solutions/app.py`. diff --git a/data-track/week-11/exercise_4/app.py b/data-track/week-11/exercise_4/app.py deleted file mode 100644 index 3d9eec3..0000000 --- a/data-track/week-11/exercise_4/app.py +++ /dev/null @@ -1,65 +0,0 @@ -""" -Exercise 4: add a daily-trend chart to the metrics app. -""" - -import os - -import pandas as pd -import psycopg2 -import streamlit as st -from dotenv import load_dotenv - -load_dotenv() - -PG_URL = os.environ["PG_URL"] -PG_SCHEMA = os.environ.get("PG_SCHEMA", "dev_yourname") - - -@st.cache_data(ttl=60) -def get_daily_trips(pg_url: str, schema: str) -> pd.DataFrame: - """Return daily trip counts for fct_trips. - - TODO: implement this function. - Use psycopg2.connect(pg_url) to connect. - Run: SELECT date_trunc('day', pickup_datetime) AS day, COUNT(*) AS trips - FROM {schema}.fct_trips - GROUP BY 1 - ORDER BY 1 - Return a DataFrame with a "day" index and a "trips" column - (e.g. pd.DataFrame(rows, columns=["day", "trips"]).set_index("day")). - """ - raise NotImplementedError("TODO: implement get_daily_trips") - - -@st.cache_data(ttl=60) -def get_fct_trips_freshness(pg_url: str, schema: str) -> dict: - with psycopg2.connect(pg_url) as conn, conn.cursor() as cur: - cur.execute( - f"SELECT COUNT(*), MAX(pickup_datetime) FROM {schema}.fct_trips" - ) - row_count, last_pickup = cur.fetchone() - return {"row_count": row_count or 0, "last_pickup": last_pickup} - - -st.title("NYC Taxi Metrics") - -# Panel 1: daily trip volume -st.subheader("Daily trip volume") -try: - daily = get_daily_trips(PG_URL, PG_SCHEMA) - st.line_chart(daily) -except NotImplementedError: - st.warning("Implement `get_daily_trips` to see the trend chart.") -except Exception as exc: - st.error(f"Postgres error: {exc}") - -# Panel 2: data freshness -st.subheader("Data freshness") -try: - freshness = get_fct_trips_freshness(PG_URL, PG_SCHEMA) - col1, col2 = st.columns(2) - col1.metric("Row count", f"{freshness['row_count']:,}") - last = freshness["last_pickup"] - col2.metric("Last pickup", str(last) if last else "None") -except Exception as exc: - st.error(f"Postgres error: {exc}") diff --git a/data-track/week-11/exercise_4/requirements.txt b/data-track/week-11/exercise_4/requirements.txt deleted file mode 100644 index b530158..0000000 --- a/data-track/week-11/exercise_4/requirements.txt +++ /dev/null @@ -1,4 +0,0 @@ -streamlit>=1.35.0 -psycopg2-binary>=2.9.9 -pandas>=2.2.0 -python-dotenv>=1.0.0 diff --git a/data-track/week-11/exercise_4/solutions/app.py b/data-track/week-11/exercise_4/solutions/app.py deleted file mode 100644 index 5d21451..0000000 --- a/data-track/week-11/exercise_4/solutions/app.py +++ /dev/null @@ -1,48 +0,0 @@ -"""Exercise 4 solution.""" - -import os - -import pandas as pd -import psycopg2 -import streamlit as st -from dotenv import load_dotenv - -load_dotenv() - -PG_URL = os.environ["PG_URL"] -PG_SCHEMA = os.environ.get("PG_SCHEMA", "dev_yourname") - - -@st.cache_data(ttl=60) -def get_daily_trips(pg_url: str, schema: str) -> pd.DataFrame: - with psycopg2.connect(pg_url) as conn, conn.cursor() as cur: - cur.execute( - f"SELECT date_trunc('day', pickup_datetime) AS day, COUNT(*) AS trips " - f"FROM {schema}.fct_trips GROUP BY 1 ORDER BY 1" - ) - rows = cur.fetchall() - return pd.DataFrame(rows, columns=["day", "trips"]).set_index("day") - - -@st.cache_data(ttl=60) -def get_fct_trips_freshness(pg_url: str, schema: str) -> dict: - with psycopg2.connect(pg_url) as conn, conn.cursor() as cur: - cur.execute( - f"SELECT COUNT(*), MAX(pickup_datetime) FROM {schema}.fct_trips" - ) - row_count, last_pickup = cur.fetchone() - return {"row_count": row_count or 0, "last_pickup": last_pickup} - - -st.title("NYC Taxi Metrics") - -st.subheader("Daily trip volume") -daily = get_daily_trips(PG_URL, PG_SCHEMA) -st.line_chart(daily) - -st.subheader("Data freshness") -freshness = get_fct_trips_freshness(PG_URL, PG_SCHEMA) -col1, col2 = st.columns(2) -col1.metric("Row count", f"{freshness['row_count']:,}") -last = freshness["last_pickup"] -col2.metric("Last pickup", str(last) if last else "None")