diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md new file mode 100644 index 0000000..0d596a3 --- /dev/null +++ b/.github/pull_request_template.md @@ -0,0 +1,32 @@ +## What I built + + + +## Why this approach + + + +## Contract impact + + + +None + +## How to run + + + +```bash + +``` + +## Self-check + +- [ ] I ran this and it works +- [ ] Tests pass locally +- [ ] No secrets, tokens, or connection strings in the diff +- [ ] This pull request does one thing diff --git a/.github/workflows/pr-checks.yml b/.github/workflows/pr-checks.yml new file mode 100644 index 0000000..d1dc344 --- /dev/null +++ b/.github/workflows/pr-checks.yml @@ -0,0 +1,114 @@ +name: PR checks + +# Two gates on every pull request: +# +# 1. The description uses the template. GitHub only auto-fills the template in +# the web "compose" form and in `gh pr create` with no --body. A pull request +# opened through the REST API or `gh pr create --body "..."`, which is the +# path most AI tools take, silently skips it. This check is the only thing +# that actually enforces it. +# +# 2. The diff stays reviewable. A two thousand line pull request does not get +# reviewed, it gets approved, which is not the same thing. Size is the real +# problem behind unreviewable AI-generated changes, so the limit is a number +# rather than an awkward conversation between teammates. +# +# Recovery is automatic for both: editing the description fires the `edited` +# event and re-runs these checks. No new commit needed. + +on: + pull_request: + types: [opened, edited, reopened, synchronize] + +permissions: + contents: read + +env: + MAX_CHANGED_LINES: 400 + +jobs: + body: + name: Description uses the template + runs-on: ubuntu-latest + steps: + - name: Check required sections are present + env: + PR_BODY: ${{ github.event.pull_request.body }} + run: | + set -euo pipefail + required=( + "## What I built" + "## Why this approach" + "## Contract impact" + "## How to run" + "## Self-check" + ) + missing=() + for section in "${required[@]}"; do + if ! printf '%s' "$PR_BODY" | grep -qiF "$section"; then + missing+=("$section") + fi + done + if [ ${#missing[@]} -ne 0 ]; then + echo "::error::Your pull request description is missing required sections. Start from .github/pull_request_template.md and keep these headings:" + for m in "${missing[@]}"; do echo " - $m"; done + echo "" + echo "Click 'Edit' on the description, paste the template, and fill it in." + echo "Editing the description re-runs this check automatically." + exit 1 + fi + echo "All required sections present." + + size: + name: Diff stays reviewable + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + with: + fetch-depth: 0 + + - name: Count changed lines, excluding generated files + id: count + env: + BASE_SHA: ${{ github.event.pull_request.base.sha }} + HEAD_SHA: ${{ github.event.pull_request.head.sha }} + run: | + set -euo pipefail + git fetch --no-tags --depth=1 origin "$BASE_SHA" 2>/dev/null || true + changed=$(git diff --numstat "$BASE_SHA" "$HEAD_SHA" -- \ + . \ + ':(exclude)**/uv.lock' \ + ':(exclude)**/package-lock.json' \ + ':(exclude)**/poetry.lock' \ + ':(exclude)**/*.lock' \ + ':(exclude)**/target/**' \ + ':(exclude)**/dbt_packages/**' \ + ':(exclude)**/node_modules/**' \ + | awk '{ add += ($1 == "-" ? 0 : $1); del += ($2 == "-" ? 0 : $2) } END { print add + del + 0 }') + echo "changed=$changed" >> "$GITHUB_OUTPUT" + echo "Changed lines, excluding generated files: $changed" + + - name: Enforce the limit, unless an override is documented + env: + PR_BODY: ${{ github.event.pull_request.body }} + CHANGED: ${{ steps.count.outputs.changed }} + run: | + set -euo pipefail + if [ "$CHANGED" -le "$MAX_CHANGED_LINES" ]; then + echo "$CHANGED changed lines is within the limit of $MAX_CHANGED_LINES." + exit 0 + fi + if printf '%s' "$PR_BODY" | grep -qiE '^[[:space:]]*Oversized:[[:space:]]*\S'; then + echo "::warning::$CHANGED changed lines exceeds $MAX_CHANGED_LINES, but an 'Oversized:' reason is documented in the description." + exit 0 + fi + echo "::error::$CHANGED changed lines exceeds the limit of $MAX_CHANGED_LINES." + echo "" + echo "Split this into smaller pull requests, one purpose each. A reviewer cannot" + echo "meaningfully check a change this size, and approving it without reading is" + echo "worse than not reviewing at all." + echo "" + echo "If it genuinely cannot be split, add a line to the description:" + echo " Oversized: " + echo "Editing the description re-runs this check." + exit 1 diff --git a/.gitignore b/.gitignore index dbdf4d3..fa9a70c 100644 --- a/.gitignore +++ b/.gitignore @@ -202,3 +202,31 @@ build/ # Docker .docker/ .buildx-cache/ + +# --- Python (data track) --- +__pycache__/ +*.py[cod] +.venv/ +venv/ +.pytest_cache/ +.ruff_cache/ +.mypy_cache/ +*.egg-info/ + +# --- dbt --- +target/ +dbt_packages/ +logs/ + +# --- Airflow (Astro) --- +# .astro/config.yaml must be committed: without it the folder is not an Astro +# project and `astro dev start` refuses to run. +.astro/config.yaml.lock +airflow_settings.yaml + +# --- Java / Spring Boot (backend track) --- +*.class +build/ +.gradle/ +.mvn/ +.user.yml diff --git a/data/.dockerignore b/data/.dockerignore new file mode 100644 index 0000000..89886f7 --- /dev/null +++ b/data/.dockerignore @@ -0,0 +1,9 @@ +.env +.venv +__pycache__/ +*.pyc +dbt/target/ +dbt/dbt_packages/ +dbt/logs/ +airflow/ +docs/ diff --git a/data/.env.example b/data/.env.example new file mode 100644 index 0000000..25c37f5 --- /dev/null +++ b/data/.env.example @@ -0,0 +1,25 @@ +# Copy to .env and fill in. Never commit .env. +# +# Local development uses the Postgres started by docker compose, so the +# defaults below work without an Azure account. Swap them for your team's +# Azure values when you deploy. + +# --- Source API ------------------------------------------------------------- +# Arbeitnow needs no key. Replace with your team's source. +SOURCE_API_URL=https://www.arbeitnow.com/api/job-board-api + +# --- Postgres --------------------------------------------------------------- +POSTGRES_HOST=localhost +POSTGRES_PORT=5432 +POSTGRES_DB=finalproject +POSTGRES_USER=postgres +POSTGRES_PASSWORD=postgres +POSTGRES_SCHEMA_RAW=raw + +# --- dbt -------------------------------------------------------------------- +# The schema dbt builds into. Use your own name so teammates do not collide. +DBT_SCHEMA=analytics + +# --- Azure Blob Storage (optional until you deploy) ------------------------- +# AZURE_STORAGE_CONNECTION_STRING= +# AZURE_STORAGE_CONTAINER=raw diff --git a/data/Dockerfile b/data/Dockerfile new file mode 100644 index 0000000..01bed4a --- /dev/null +++ b/data/Dockerfile @@ -0,0 +1,23 @@ +# Container image for the ingestion pipeline. +# +# Build: docker build -t final-project-data . +# Run: docker run --rm --env-file .env final-project-data +# +# This is the image you push to Azure Container Registry and run as a +# Container Apps Job, exactly as in Week 6. +FROM python:3.11-slim + +WORKDIR /app + +# Dependencies are copied first so Docker can cache the install layer. Change +# your source code and the rebuild stays fast; change dependencies and it does not. +COPY pyproject.toml ./ +RUN pip install --no-cache-dir \ + "requests>=2.32.0" \ + "pydantic>=2.9.0" \ + "psycopg[binary]>=3.2.0" \ + "python-dotenv>=1.0.0" + +COPY src/ ./src/ + +CMD ["python", "-m", "src.pipeline"] diff --git a/data/README.md b/data/README.md index a2a5d5f..472b87b 100644 --- a/data/README.md +++ b/data/README.md @@ -1 +1,88 @@ -# Final Project Data Pipeline \ No newline at end of file +# Final Project Data Pipeline + +Starter code for the data half of the final project: fetch data from a source, +validate it, store it, shape it with dbt, and publish a mart the backend team +reads. It runs end to end the moment you clone it, against a local Postgres and +a public API that needs no key, so your first hour goes into your product +rather than into setup. + +## Run it in five minutes + +```bash +cd data +cp .env.example .env +docker compose up -d db # local Postgres on :5432 + +uv venv && uv pip install -e ".[dbt]" +uv run python -m src.pipeline # fetch, validate, store + +cd dbt && uv run dbt build --profiles-dir . +``` + +To run the pipeline the way Azure will run it, in the container: + +```bash +docker compose run --rm pipeline +``` + +> Inside a container, `localhost` is the container itself, not your machine. +> That is why the `pipeline` service overrides `POSTGRES_HOST` to `postgres`, +> the service name on the compose network. Plain +> `docker run --env-file .env` cannot reach your local database. + +You should see around 175 rows land in `raw.postings`, then `stg_postings` and +`fct_postings` build with all tests passing. Run the pipeline twice: the row +count stays the same, because writes are upserts. + +## What is here + +| Path | What it does | +|---|---| +| `src/config.py` | Reads every setting from environment variables and fails loudly when one is missing | +| `src/models.py` | Pydantic validation for incoming records | +| `src/ingest.py` | Calls the source API, validates, counts rejects | +| `src/storage.py` | Creates the raw schema and upserts rows | +| `src/pipeline.py` | Entry point, wires the three steps together | +| `dbt/models/staging/` | Cleans and renames. No business logic | +| `dbt/models/marts/fct_postings.sql` | **The contract with the backend team** | +| `dbt/tests/` | Two custom tests, including a zero-row check | +| `airflow/dags/pipeline_dag.py` | Daily schedule: ingest, then dbt build | +| `Dockerfile` | The image you push to Azure Container Registry | +| `optional/` | Bicep, Databricks, and Streamlit modules. None required | + +## Making it yours + +The template ships a job-postings example so it runs immediately. Swapping in +your team's data source is four edits: + +1. `.env`: point `SOURCE_API_URL` at your source. +2. `src/models.py`: change the Pydantic model to match your records. +3. `src/storage.py`: change the table definition and upsert to match. +4. `dbt/models/`: rename the models and columns to your domain. + +Do this in your first two days. Everything after that builds on the shape you +choose here. + +> Verify your source before you commit to it: call it once, print a record, and +> confirm you can parse it. An idea you love with a source you cannot reach is +> worth less than a plain idea that works. + +## The mart is a contract + +`fct_postings` is what the backend reads to build endpoints. Adding a column is +safe. Renaming or removing one breaks the backend, so agree it with them first +and change it in both places at once. + +Every column is documented in `dbt/models/marts/_fct_postings.yml`. Hand that +file to the backend trainees on day one and they can write endpoints before +your pipeline is finished. See `docs/mart_contract.md` for how to work on it +together. + +## Secrets + +No credentials live in this folder. `dbt/profiles.yml` is committed on purpose: +every value in it comes from `env_var(...)`, so it holds nothing secret. Real +values live in `.env`, which is git-ignored, and in your deployment environment. + +Never commit `.env`, and never paste a connection string into a chat message or +an LLM prompt. diff --git a/data/airflow/.astro/config.yaml b/data/airflow/.astro/config.yaml new file mode 100644 index 0000000..74ee750 --- /dev/null +++ b/data/airflow/.astro/config.yaml @@ -0,0 +1,2 @@ +project: + name: final-project-data diff --git a/data/airflow/.env.example b/data/airflow/.env.example new file mode 100644 index 0000000..b034c71 --- /dev/null +++ b/data/airflow/.env.example @@ -0,0 +1,19 @@ +# Copy to .env before running `astro dev start`. Never commit .env. +# +# POSTGRES_HOST is the compose service name "db", not localhost and not +# "postgres": Airflow's own stack has a service called postgres, and that name +# would resolve to Airflow's metadata database instead of yours. +# the Airflow +# containers join the "finalproject" network created by ../docker-compose.yml. +# Start the database first with: (cd .. && docker compose up -d postgres) + +SOURCE_API_URL=https://www.arbeitnow.com/api/job-board-api + +POSTGRES_HOST=db +POSTGRES_PORT=5432 +POSTGRES_DB=finalproject +POSTGRES_USER=postgres +POSTGRES_PASSWORD=postgres +POSTGRES_SCHEMA_RAW=raw + +DBT_SCHEMA=analytics diff --git a/data/airflow/.gitignore b/data/airflow/.gitignore new file mode 100644 index 0000000..a0ab42c --- /dev/null +++ b/data/airflow/.gitignore @@ -0,0 +1,3 @@ +.astro/config.yaml.lock +airflow_settings.yaml +logs/ diff --git a/data/airflow/Dockerfile b/data/airflow/Dockerfile new file mode 100644 index 0000000..1eb4744 --- /dev/null +++ b/data/airflow/Dockerfile @@ -0,0 +1,5 @@ +# Astro runtime, the same image family you used in Week 12. +# +# Start locally: astro dev start +# Airflow UI: http://localhost:8080 +FROM astrocrpublic.azurecr.io/runtime:3.3-2 diff --git a/data/airflow/dags/pipeline_dag.py b/data/airflow/dags/pipeline_dag.py new file mode 100644 index 0000000..c01259f --- /dev/null +++ b/data/airflow/dags/pipeline_dag.py @@ -0,0 +1,51 @@ +"""Daily orchestration for the final project pipeline. + +Two tasks in sequence: ingest raw data, then let dbt shape it. Keeping them +separate means a dbt failure does not force you to re-fetch from the API, and +you can see at a glance which half broke. + +Set these Airflow variables (or environment variables) before the first run: + SOURCE_API_URL, POSTGRES_HOST, POSTGRES_PORT, POSTGRES_DB, + POSTGRES_USER, POSTGRES_PASSWORD, POSTGRES_SCHEMA_RAW, DBT_SCHEMA +""" + +from __future__ import annotations + +from datetime import datetime, timedelta + +from airflow.providers.standard.operators.bash import BashOperator +from airflow.sdk import dag, task + +DEFAULT_ARGS = { + "owner": "data-team", + "retries": 2, + "retry_delay": timedelta(minutes=5), +} + + +@dag( + dag_id="final_project_pipeline", + description="Ingest source data, then build dbt models", + start_date=datetime(2026, 1, 1), + schedule="0 6 * * *", + catchup=False, + default_args=DEFAULT_ARGS, + tags=["final-project"], +) +def final_project_pipeline(): + @task + def ingest() -> int: + """Fetch, validate, and store raw records.""" + from src.pipeline import run + + return run() + + dbt_build = BashOperator( + task_id="dbt_build", + bash_command="cd /usr/local/airflow/include/dbt && dbt build --profiles-dir .", + ) + + ingest() >> dbt_build + + +final_project_pipeline() diff --git a/data/airflow/docker-compose.override.yml b/data/airflow/docker-compose.override.yml new file mode 100644 index 0000000..f4fe0d5 --- /dev/null +++ b/data/airflow/docker-compose.override.yml @@ -0,0 +1,38 @@ +# Local development only. Astro merges this into its own compose stack. +# +# Two things happen here: +# +# 1. The pipeline code and the dbt project live one level up, next to this +# folder, so there is exactly one copy of each. The bind mounts make them +# visible inside the Airflow containers rather than duplicating them. +# +# 2. The containers join the "finalproject" network created by +# ../docker-compose.yml, so they reach the database as "db" on port +# 5432. Going through host.docker.internal instead would break on any +# machine that already runs something on port 5432. +# +# Start the database first: (cd .. && docker compose up -d db) +# When you deploy, bake these paths into the image with COPY. +x-local: &local + volumes: + - ../src:/usr/local/airflow/src:ro + - ../dbt:/usr/local/airflow/include/dbt:rw + environment: + PYTHONPATH: /usr/local/airflow + networks: + - default + - finalproject + +services: + scheduler: + <<: *local + dag-processor: + <<: *local + api-server: + <<: *local + triggerer: + <<: *local + +networks: + finalproject: + external: true diff --git a/data/airflow/packages.txt b/data/airflow/packages.txt new file mode 100644 index 0000000..9e59294 --- /dev/null +++ b/data/airflow/packages.txt @@ -0,0 +1,2 @@ +# OS-level packages installed into the Airflow image (one per line). +# Python packages belong in requirements.txt instead. diff --git a/data/airflow/requirements.txt b/data/airflow/requirements.txt new file mode 100644 index 0000000..a7d4e6f --- /dev/null +++ b/data/airflow/requirements.txt @@ -0,0 +1,8 @@ +# Python packages available to your DAGs. +# The Astro runtime already ships Airflow itself. +dbt-core>=1.8.0 +dbt-postgres>=1.8.0 +psycopg[binary]>=3.2.0 +pydantic>=2.9.0 +requests>=2.32.0 +python-dotenv>=1.0.0 diff --git a/data/dbt/dbt_project.yml b/data/dbt/dbt_project.yml new file mode 100644 index 0000000..8a8d83a --- /dev/null +++ b/data/dbt/dbt_project.yml @@ -0,0 +1,19 @@ +name: 'final_project' +version: '1.0.0' +profile: 'final_project' + +model-paths: ["models"] +test-paths: ["tests"] +seed-paths: ["seeds"] +macro-paths: ["macros"] + +clean-targets: + - "target" + - "dbt_packages" + +models: + final_project: + staging: + +materialized: view + marts: + +materialized: table diff --git a/data/dbt/models/marts/_fct_postings.yml b/data/dbt/models/marts/_fct_postings.yml new file mode 100644 index 0000000..fe4995b --- /dev/null +++ b/data/dbt/models/marts/_fct_postings.yml @@ -0,0 +1,34 @@ +version: 2 + +models: + - name: fct_postings + description: > + One row per posting. This is the published contract consumed by the + backend API. Adding a column is safe; renaming or removing one breaks + the backend and must be agreed with them first. + columns: + - name: posting_id + description: Stable unique identifier. Use this as the API resource id. + tests: [unique, not_null] + - name: title + description: Job title as advertised. + tests: [not_null] + - name: company_name + description: Hiring company name. + tests: [not_null] + - name: location + description: Free-text location. Null when the posting gives none. + - name: is_remote + description: True when the posting is advertised as remote. + tests: [not_null] + - name: tags + description: JSON array of tags from the source. Query with Postgres JSONB operators. + - name: posted_at + description: Timestamp the posting was published at the source. + tests: [not_null] + - name: posted_date + description: Date part of posted_at. Group by this for daily counts. + tests: [not_null] + - name: ingested_at + description: When your pipeline last saw this record. Use it to show data freshness in the UI. + tests: [not_null] diff --git a/data/dbt/models/marts/fct_postings.sql b/data/dbt/models/marts/fct_postings.sql new file mode 100644 index 0000000..6c5586b --- /dev/null +++ b/data/dbt/models/marts/fct_postings.sql @@ -0,0 +1,22 @@ +-- This mart is the contract with the backend team. +-- +-- Its columns are what backend/ reads to build API endpoints, so treat a +-- change here the way you would treat changing a public API: agree it with +-- the backend trainees first, then change it in both places. +with postings as ( + + select * from {{ ref('stg_postings') }} + +) + +select + posting_id, + title, + company_name, + location, + is_remote, + tags, + posted_at, + ingested_at, + date(posted_at) as posted_date +from postings diff --git a/data/dbt/models/staging/_sources.yml b/data/dbt/models/staging/_sources.yml new file mode 100644 index 0000000..39f4e89 --- /dev/null +++ b/data/dbt/models/staging/_sources.yml @@ -0,0 +1,13 @@ +version: 2 + +sources: + - name: raw + description: Tables written by the ingestion pipeline in src/. + schema: "{{ env_var('POSTGRES_SCHEMA_RAW', 'raw') }}" + tables: + - name: postings + description: One row per posting as received from the source API. + loaded_at_field: ingested_at + freshness: + warn_after: {count: 24, period: hour} + error_after: {count: 48, period: hour} diff --git a/data/dbt/models/staging/_stg_postings.yml b/data/dbt/models/staging/_stg_postings.yml new file mode 100644 index 0000000..1aa93b3 --- /dev/null +++ b/data/dbt/models/staging/_stg_postings.yml @@ -0,0 +1,21 @@ +version: 2 + +models: + - name: stg_postings + description: Cleaned postings, one row per posting. + columns: + - name: posting_id + description: Unique identifier from the source system. + tests: [unique, not_null] + - name: title + description: Job title as advertised. + tests: [not_null] + - name: company_name + description: Hiring company. + tests: [not_null] + - name: is_remote + description: Whether the posting is advertised as remote. + tests: [not_null] + - name: posted_at + description: When the posting was published at the source. + tests: [not_null] diff --git a/data/dbt/models/staging/stg_postings.sql b/data/dbt/models/staging/stg_postings.sql new file mode 100644 index 0000000..6b04d82 --- /dev/null +++ b/data/dbt/models/staging/stg_postings.sql @@ -0,0 +1,23 @@ +-- Staging does one job: clean and rename. No business logic lives here. +with source as ( + + select * from {{ source('raw', 'postings') }} + +), + +renamed as ( + + select + slug as posting_id, + trim(title) as title, + trim(company_name) as company_name, + nullif(trim(location), '') as location, + remote as is_remote, + tags as tags, + created_at as posted_at, + ingested_at as ingested_at + from source + +) + +select * from renamed diff --git a/data/dbt/profiles.yml b/data/dbt/profiles.yml new file mode 100644 index 0000000..453f4d4 --- /dev/null +++ b/data/dbt/profiles.yml @@ -0,0 +1,14 @@ +# Every value comes from an environment variable, so this file holds no +# secrets and is safe to commit. Set them in .env (see ../.env.example). +final_project: + target: dev + outputs: + dev: + type: postgres + host: "{{ env_var('POSTGRES_HOST') }}" + port: "{{ env_var('POSTGRES_PORT', '5432') | int }}" + dbname: "{{ env_var('POSTGRES_DB') }}" + user: "{{ env_var('POSTGRES_USER') }}" + password: "{{ env_var('POSTGRES_PASSWORD') }}" + schema: "{{ env_var('DBT_SCHEMA') }}" + threads: 4 diff --git a/data/dbt/tests/assert_posted_at_not_in_future.sql b/data/dbt/tests/assert_posted_at_not_in_future.sql new file mode 100644 index 0000000..f1472a7 --- /dev/null +++ b/data/dbt/tests/assert_posted_at_not_in_future.sql @@ -0,0 +1,5 @@ +-- A posting dated in the future means the source changed its date format or +-- your parsing is wrong. Either way you want to know before the backend does. +select posting_id, posted_at +from {{ ref('fct_postings') }} +where posted_at > now() + interval '1 day' diff --git a/data/dbt/tests/assert_postings_not_empty.sql b/data/dbt/tests/assert_postings_not_empty.sql new file mode 100644 index 0000000..652d6d8 --- /dev/null +++ b/data/dbt/tests/assert_postings_not_empty.sql @@ -0,0 +1,5 @@ +-- An empty mart passes every column test, which is exactly why it needs its +-- own check. A silent zero-row build is the failure students hit most often. +select 1 as problem +from (select count(*) as n from {{ ref('fct_postings') }}) counted +where counted.n = 0 diff --git a/data/docker-compose.yml b/data/docker-compose.yml new file mode 100644 index 0000000..fffb36a --- /dev/null +++ b/data/docker-compose.yml @@ -0,0 +1,55 @@ +# Local development stack. +# +# Two ways to run the pipeline, and the difference matters: +# +# From your machine (fast to iterate, uses POSTGRES_HOST=localhost): +# docker compose up -d db +# uv run python -m src.pipeline +# cd dbt && uv run dbt build --profiles-dir . +# +# In the container, the way Azure will run it (uses POSTGRES_HOST=db): +# docker compose run --rm pipeline +# +# The service is called "db" rather than "postgres" on purpose: the Airflow +# stack ships its own service named postgres, and a name clash there sends +# your DAG to Airflow's metadata database instead of your data. +# +# Inside a container, "localhost" is the container itself, not your machine. +# That is why the pipeline service overrides POSTGRES_HOST below instead of +# taking it from .env. +services: + db: + image: postgres:16 + environment: + POSTGRES_DB: finalproject + POSTGRES_USER: postgres + POSTGRES_PASSWORD: postgres + ports: + - "5432:5432" + volumes: + - pgdata:/var/lib/postgresql/data + healthcheck: + test: ["CMD-SHELL", "pg_isready -U postgres"] + interval: 5s + timeout: 5s + retries: 10 + + pipeline: + build: . + env_file: .env + environment: + # Reach Postgres by its service name on the compose network. + POSTGRES_HOST: db + depends_on: + db: + condition: service_healthy + +volumes: + pgdata: + +# A fixed network name so the Airflow stack, which is a separate compose +# project, can attach to it. Without a fixed name Docker derives one from the +# folder name, which differs per checkout. +networks: + default: + name: finalproject diff --git a/data/docs/mart_contract.md b/data/docs/mart_contract.md new file mode 100644 index 0000000..01c6271 --- /dev/null +++ b/data/docs/mart_contract.md @@ -0,0 +1,56 @@ +# The mart contract + +This is the agreement between the data trainees and the backend trainees. It +exists because the frontend trainee cannot start until the API shape is known, +and the API shape cannot be known until the mart shape is. + +Write it in week one, before anyone builds anything. + +## What a contract is + +A published dbt mart plus its `.yml` file. The `.yml` names every column and +says what it means. `dbt/models/marts/_fct_postings.yml` is a worked example. + +That file is the whole contract. Once it exists, the backend can write +endpoints against columns that do not have data in them yet, and the frontend +can build screens against endpoints that return fixtures. + +## How to agree it + +1. **Data trainees** draft the mart columns from what the product needs, not + from what the source happens to provide. +2. **Backend trainees** review it against the endpoints they plan to expose. A + column nobody will serve is a column you do not need to build. +3. Both pairs sign off, then it goes in the repository. + +## Changing it later + +You will change it. That is fine, as long as it is deliberate. + +| Change | Safe? | What to do | +|---|---|---| +| Add a column | Yes | Tell the backend it exists | +| Add a test | Yes | Just do it | +| Rename a column | No | Agree first, change both sides in the same day | +| Remove a column | No | Confirm no endpoint reads it, then remove | +| Change a type | No | Agree first. This breaks deserialisation quietly | + +The rule of thumb: if the backend would have to change code, it is not a +unilateral change. + +## Serving the mart + +The backend reads the mart directly from Postgres. It does not re-implement +the transformations, and the data pipeline does not expose HTTP endpoints. +Each side does one job. + +If the backend needs a shape the mart does not have, the answer is a new mart +model, not a join written in Java. Business logic lives in dbt, where it is +tested and documented. + +## Freshness + +`ingested_at` tells you when the pipeline last saw a record. Surface it in the +UI, for example "updated 20 minutes ago". Users trust a number with a +timestamp far more than a number without one, and it makes a stale pipeline +visible during the demo instead of invisible. diff --git a/data/optional/README.md b/data/optional/README.md new file mode 100644 index 0000000..3e107fc --- /dev/null +++ b/data/optional/README.md @@ -0,0 +1,11 @@ +# Optional modules + +Nothing in this folder is required. Week 15 asks for a working pipeline, not for +every tool you have seen. Add a module only when your team has the required +pipeline running and wants to go further. + +| Folder | Adds | Data Track week | +|---|---|---| +| `bicep/` | Provision your Azure resources from code instead of clicking in the portal | 14 | +| `databricks/` | Run transformations on Databricks instead of Postgres | 13 | +| `streamlit/` | An operations dashboard showing pipeline health | 11 | diff --git a/data/optional/bicep/main.bicep b/data/optional/bicep/main.bicep new file mode 100644 index 0000000..85e769e --- /dev/null +++ b/data/optional/bicep/main.bicep @@ -0,0 +1,21 @@ +// Provisions the Azure resources this project needs. +// +// Deploy: +// az deployment group create \ +// --resource-group \ +// --template-file main.bicep \ +// --parameters projectName= +// +// param = an input you pass at deploy time. +param location string = resourceGroup().location +param projectName string + +module storage 'modules/storage.bicep' = { + name: 'storageDeploy' + params: { + location: location + storageName: 'st${toLower(projectName)}' + } +} + +output storageId string = storage.outputs.storageId diff --git a/data/optional/bicep/modules/storage.bicep b/data/optional/bicep/modules/storage.bicep new file mode 100644 index 0000000..205c4b8 --- /dev/null +++ b/data/optional/bicep/modules/storage.bicep @@ -0,0 +1,18 @@ +// A storage account for raw files landed by the pipeline. +param location string +param storageName string + +resource storage 'Microsoft.Storage/storageAccounts@2023-01-01' = { + name: storageName + location: location + sku: { + name: 'Standard_LRS' + } + kind: 'StorageV2' + properties: { + minimumTlsVersion: 'TLS1_2' + allowBlobPublicAccess: false + } +} + +output storageId string = storage.id diff --git a/data/optional/databricks/README.md b/data/optional/databricks/README.md new file mode 100644 index 0000000..be410bb --- /dev/null +++ b/data/optional/databricks/README.md @@ -0,0 +1,24 @@ +# Databricks module + +Use this only if your team has a reason to move transformations off Postgres, +for example a dataset too large to model comfortably in it. + +## Switching dbt to Databricks + +Add a second output to `../../dbt/profiles.yml`: + +```yaml + databricks: + type: databricks + catalog: "{{ env_var('DATABRICKS_CATALOG') }}" + schema: "{{ env_var('DBT_SCHEMA') }}" + host: "{{ env_var('DATABRICKS_HOST') }}" + http_path: "{{ env_var('DATABRICKS_HTTP_PATH') }}" + token: "{{ env_var('DATABRICKS_TOKEN') }}" + threads: 4 +``` + +Then run `dbt build --target databricks`. Your models stay the same, which is +the point of keeping business logic in dbt rather than in notebooks. + +Install the adapter with `uv pip install dbt-databricks`. diff --git a/data/optional/streamlit/app.py b/data/optional/streamlit/app.py new file mode 100644 index 0000000..c77ee0c --- /dev/null +++ b/data/optional/streamlit/app.py @@ -0,0 +1,45 @@ +"""Operations dashboard for the pipeline. + +This is for your team, not for end users. The product UI is the frontend +trainee's job. This page answers one question: is the pipeline healthy? + +Run: uv run streamlit run optional/streamlit/app.py +""" + +import os + +import pandas as pd +import psycopg +import streamlit as st + +st.set_page_config(page_title="Pipeline health", page_icon="📊") +st.title("Pipeline health") + +DSN = ( + f"host={os.environ['POSTGRES_HOST']} port={os.getenv('POSTGRES_PORT', '5432')} " + f"dbname={os.environ['POSTGRES_DB']} user={os.environ['POSTGRES_USER']} " + f"password={os.environ['POSTGRES_PASSWORD']}" +) +SCHEMA = os.getenv("DBT_SCHEMA", "analytics") + + +@st.cache_data(ttl=60) +def load_freshness() -> pd.DataFrame: + query = f""" + select + max(ingested_at) as last_ingested, + count(*) as row_count, + count(distinct posted_date) as days_covered + from {SCHEMA}.fct_postings + """ + with psycopg.connect(DSN) as conn: + return pd.read_sql(query, conn) + + +stats = load_freshness() +col1, col2, col3 = st.columns(3) +col1.metric("Rows", int(stats["row_count"][0])) +col2.metric("Days covered", int(stats["days_covered"][0])) +col3.metric("Last ingest", str(stats["last_ingested"][0])) + +st.caption("Add a chart per metric your team cares about. Keep it to what you would check at 9am.") diff --git a/data/pyproject.toml b/data/pyproject.toml new file mode 100644 index 0000000..40f1ca3 --- /dev/null +++ b/data/pyproject.toml @@ -0,0 +1,25 @@ +[project] +name = "final-project-data" +version = "0.1.0" +description = "Data pipeline for the HYF final project" +requires-python = ">=3.11" +dependencies = [ + "requests>=2.32.0", + "pydantic>=2.9.0", + "psycopg[binary]>=3.2.0", + "python-dotenv>=1.0.0", +] + +[project.optional-dependencies] +dbt = ["dbt-core>=1.8.0", "dbt-postgres>=1.8.0"] +dev = ["pytest>=8.3.0", "ruff>=0.6.0"] + +[tool.ruff] +line-length = 100 + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["src"] diff --git a/data/src/__init__.py b/data/src/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/data/src/config.py b/data/src/config.py new file mode 100644 index 0000000..2c16d3c --- /dev/null +++ b/data/src/config.py @@ -0,0 +1,56 @@ +"""Configuration read from environment variables. + +Every setting comes from the environment. Nothing is hard-coded and no +secret is ever committed, which is the same rule you followed from Week 2 +onwards. Copy .env.example to .env for local development. +""" + +import os +from dataclasses import dataclass + +from dotenv import load_dotenv + +load_dotenv() + + +@dataclass(frozen=True) +class Config: + source_api_url: str + postgres_host: str + postgres_port: int + postgres_db: str + postgres_user: str + postgres_password: str + raw_schema: str + + @property + def postgres_dsn(self) -> str: + return ( + f"host={self.postgres_host} port={self.postgres_port} " + f"dbname={self.postgres_db} user={self.postgres_user} " + f"password={self.postgres_password}" + ) + + +def _required(name: str) -> str: + value = os.getenv(name) + if not value: + raise RuntimeError(f"Missing required environment variable: {name}") + return value + + +def load_config() -> Config: + """Build a Config, failing loudly when something is missing. + + Failing at startup is deliberate. A pipeline that starts with half its + configuration and dies twenty minutes later is far harder to debug. + """ + return Config( + source_api_url=_required("SOURCE_API_URL"), + postgres_host=_required("POSTGRES_HOST"), + postgres_port=int(os.getenv("POSTGRES_PORT", "5432")), + postgres_db=_required("POSTGRES_DB"), + postgres_user=_required("POSTGRES_USER"), + postgres_password=_required("POSTGRES_PASSWORD"), + raw_schema=os.getenv("POSTGRES_SCHEMA_RAW", "raw"), + ) diff --git a/data/src/ingest.py b/data/src/ingest.py new file mode 100644 index 0000000..1e21fae --- /dev/null +++ b/data/src/ingest.py @@ -0,0 +1,50 @@ +"""Fetch records from the source API and validate them. + +The default source is the Arbeitnow job board, which needs no API key so the +template runs the moment you clone it. Point SOURCE_API_URL at your team's +source and rewrite `parse_records` to match its shape. +""" + +import logging + +import requests +from pydantic import ValidationError + +from .models import Posting + +logger = logging.getLogger(__name__) + +REQUEST_TIMEOUT_SECONDS = 30 + + +def fetch_raw(url: str) -> list[dict]: + """Call the source API and return its raw records. + + Any non-2xx response raises, so a broken source fails the pipeline run + instead of silently writing zero rows. + """ + logger.info("Fetching %s", url) + response = requests.get(url, timeout=REQUEST_TIMEOUT_SECONDS) + response.raise_for_status() + payload = response.json() + records = payload.get("data", payload) + logger.info("Received %d record(s)", len(records)) + return records + + +def parse_records(records: list[dict]) -> tuple[list[Posting], int]: + """Validate raw records, returning the good ones and a rejected count. + + One malformed record should not lose you the whole batch, so invalid rows + are counted and skipped rather than raised. + """ + parsed: list[Posting] = [] + rejected = 0 + for record in records: + try: + parsed.append(Posting.model_validate(record)) + except ValidationError as exc: + rejected += 1 + logger.warning("Rejected record %s: %s", record.get("slug", ""), exc.error_count()) + logger.info("Parsed %d record(s), rejected %d", len(parsed), rejected) + return parsed, rejected diff --git a/data/src/models.py b/data/src/models.py new file mode 100644 index 0000000..9507a14 --- /dev/null +++ b/data/src/models.py @@ -0,0 +1,34 @@ +"""Validation models for the source data. + +Validating at the edge means bad records are caught where they enter the +pipeline, not three transformations later when the error message no longer +tells you anything useful. + +Replace this model with one that matches your team's data source. +""" + +from datetime import datetime + +from pydantic import BaseModel, Field, field_validator + + +class Posting(BaseModel): + """One job posting from the source API.""" + + slug: str + title: str + company_name: str = Field(alias="company_name") + location: str | None = None + remote: bool = False + tags: list[str] = Field(default_factory=list) + created_at: datetime + + @field_validator("created_at", mode="before") + @classmethod + def _epoch_to_datetime(cls, value: object) -> object: + """The source sends a Unix timestamp; store a real datetime.""" + if isinstance(value, int): + return datetime.fromtimestamp(value) + return value + + model_config = {"populate_by_name": True} diff --git a/data/src/pipeline.py b/data/src/pipeline.py new file mode 100644 index 0000000..1fca3f7 --- /dev/null +++ b/data/src/pipeline.py @@ -0,0 +1,46 @@ +"""Pipeline entry point: fetch, validate, store. + +Run locally: + docker compose up -d + uv run python -m src.pipeline + +The same module is what the container image runs, so what you test locally is +what Airflow and Azure execute. +""" + +import logging +import sys + +from .config import load_config +from .ingest import fetch_raw, parse_records +from .storage import ensure_schema, write_postings + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(name)s %(message)s", +) +logger = logging.getLogger("pipeline") + + +def run() -> int: + """Run one pipeline execution and return the number of rows written.""" + config = load_config() + + records = fetch_raw(config.source_api_url) + postings, rejected = parse_records(records) + if rejected and not postings: + raise RuntimeError("Every record failed validation: check the source shape") + + ensure_schema(config.postgres_dsn, config.raw_schema) + written = write_postings(config.postgres_dsn, config.raw_schema, postings) + + logger.info("Pipeline finished: %d row(s) written, %d rejected", written, rejected) + return written + + +if __name__ == "__main__": + try: + run() + except Exception: + logger.exception("Pipeline failed") + sys.exit(1) diff --git a/data/src/storage.py b/data/src/storage.py new file mode 100644 index 0000000..f5f8ef5 --- /dev/null +++ b/data/src/storage.py @@ -0,0 +1,83 @@ +"""Write validated records into Postgres. + +The pipeline owns the raw layer only. Everything downstream of `raw` is dbt's +job, which keeps the boundary between "getting data in" and "shaping data" +clear enough that two people can work on them at once. +""" + +import json +import logging + +import psycopg + +from .models import Posting + +logger = logging.getLogger(__name__) + +CREATE_SCHEMA = "CREATE SCHEMA IF NOT EXISTS {schema}" + +CREATE_TABLE = """ +CREATE TABLE IF NOT EXISTS {schema}.postings ( + slug TEXT PRIMARY KEY, + title TEXT NOT NULL, + company_name TEXT NOT NULL, + location TEXT, + remote BOOLEAN NOT NULL DEFAULT FALSE, + tags JSONB NOT NULL DEFAULT '[]'::jsonb, + created_at TIMESTAMP NOT NULL, + ingested_at TIMESTAMP NOT NULL DEFAULT NOW() +) +""" + +UPSERT = """ +INSERT INTO {schema}.postings + (slug, title, company_name, location, remote, tags, created_at) +VALUES (%s, %s, %s, %s, %s, %s, %s) +ON CONFLICT (slug) DO UPDATE SET + title = EXCLUDED.title, + company_name = EXCLUDED.company_name, + location = EXCLUDED.location, + remote = EXCLUDED.remote, + tags = EXCLUDED.tags, + created_at = EXCLUDED.created_at, + ingested_at = NOW() +""" + + +def ensure_schema(dsn: str, schema: str) -> None: + """Create the raw schema and table when they do not exist yet.""" + with psycopg.connect(dsn) as conn, conn.cursor() as cur: + cur.execute(CREATE_SCHEMA.format(schema=schema)) + cur.execute(CREATE_TABLE.format(schema=schema)) + conn.commit() + logger.info("Ensured %s.postings exists", schema) + + +def write_postings(dsn: str, schema: str, postings: list[Posting]) -> int: + """Upsert postings and return how many rows were written. + + Upserting rather than inserting makes the pipeline safe to re-run. Running + it twice on the same day must not double your row count, and Airflow will + re-run tasks whenever one fails. + """ + if not postings: + logger.warning("Nothing to write") + return 0 + + rows = [ + ( + p.slug, + p.title, + p.company_name, + p.location, + p.remote, + json.dumps(p.tags), + p.created_at, + ) + for p in postings + ] + with psycopg.connect(dsn) as conn, conn.cursor() as cur: + cur.executemany(UPSERT.format(schema=schema), rows) + conn.commit() + logger.info("Wrote %d row(s) into %s.postings", len(rows), schema) + return len(rows)