diff --git a/.env.example b/.env.example index e7d387e..42aa982 100644 --- a/.env.example +++ b/.env.example @@ -8,4 +8,9 @@ # echo "POSTGRES_URL=$(az keyvault secret show --vault-name kv-hyf-data --name postgres-url --query value -o tsv)" >> .env # echo "AZURE_STORAGE_CONNECTION_STRING=$(az keyvault secret show --vault-name kv-hyf-data --name storage-connection-string --query value -o tsv)" >> .env +POSTGRES_URL=postgresql://:@.postgres.database.azure.com:5432/?sslmode=require +AZURE_STORAGE_CONNECTION_STRING=DefaultEndpointsProtocol=https;AccountName=...;AccountKey=...;EndpointSuffix=core.windows.net +# Your personal schema on the shared Postgres server — replace with your GitHub handle. +# All tables are created here so they don't collide with other students. +DB_SCHEMA=dev_ LOG_LEVEL=INFO diff --git a/README.md b/README.md index 3d2bad5..d86d31d 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,7 @@ ## Architecture ```text -[Your API] ──► pipeline.py ──► Pydantic validation ──► Postgres INSERT +[Your API] ──► pipeline.py ──► Pydantic validation ──► Postgres INSERT (your schema) ──► Blob Storage (raw JSON) ``` @@ -18,6 +18,8 @@ cp .env.example .env echo "POSTGRES_URL=$(az keyvault secret show --vault-name kv-hyf-data --name postgres-url --query value -o tsv)" >> .env echo "AZURE_STORAGE_CONNECTION_STRING=$(az keyvault secret show --vault-name kv-hyf-data --name storage-connection-string --query value -o tsv)" >> .env +# Set your personal schema (replace alice with your GitHub handle): +echo "DB_SCHEMA=dev_alice" >> .env # 2. Install dependencies uv sync @@ -57,12 +59,26 @@ az containerapp job create \ --env-vars \ POSTGRES_URL="$(az keyvault secret show --vault-name kv-hyf-data --name postgres-url --query value -o tsv)" \ AZURE_STORAGE_CONNECTION_STRING="$(az keyvault secret show --vault-name kv-hyf-data --name storage-connection-string --query value -o tsv)" \ + DB_SCHEMA=dev_alice \ LOG_LEVEL=INFO # Trigger a manual run for testing (without waiting for the schedule) az containerapp job start --name my-pipeline-job --resource-group rg-hyf-data ``` +## Enable ACR push from CI (optional) + +The `push-to-acr` job in `.github/workflows/ci.yml` is commented out by default. +To enable it, add two secrets in your repo's **Settings → Secrets and variables → Actions**: + +| Secret name | Value | +|-------------|-------| +| `ACR_USERNAME` | `hyfregistry` | +| `ACR_PASSWORD` | Ask your teacher for the ACR password | + +Then uncomment the `push-to-acr` job in `ci.yml`. Every push to `main` will build +and push the image automatically. + ## Install psql `psql` is the Postgres command-line client used to verify results. Install it once: @@ -88,8 +104,8 @@ Download and run the installer from [postgresql.org/download/windows](https://ww # Check job execution az containerapp job execution list --name my-pipeline-job --resource-group rg-hyf-data --output table -# Check Postgres (set POSTGRES_URL first — see Run locally above) -psql "$POSTGRES_URL" -c "SELECT COUNT(*) FROM your_table_name;" # replace with your table name +# Check Postgres (replace dev_alice with your schema, with your table name) +psql "$POSTGRES_URL" -c "SELECT COUNT(*) FROM dev_alice.;" # Check Blob Storage az storage blob list --account-name hyfstoragedev --container-name raw --prefix pipeline/ --output table diff --git a/src/storage.py b/src/storage.py index b709550..c60f535 100644 --- a/src/storage.py +++ b/src/storage.py @@ -4,7 +4,7 @@ import logging import os from contextlib import closing -from datetime import datetime +from datetime import datetime, timezone import pandas as pd import psycopg2 @@ -15,34 +15,48 @@ def insert_readings(df: pd.DataFrame) -> None: - """Insert a DataFrame of readings into Postgres.""" + """Insert a DataFrame of readings into Postgres. + + Creates the table in your personal schema (DB_SCHEMA env var, e.g. dev_alice). + All CREATE TABLE and INSERT statements run inside that schema so your tables + never collide with other students on the shared server. + """ db_url = os.environ["POSTGRES_URL"] + schema = os.environ.get("DB_SCHEMA", "public") with closing(psycopg2.connect(db_url)) as conn: - cur = conn.cursor() - - # TODO: Replace 'weather_readings' with a unique name (e.g. alice_weather_readings) - # to avoid collisions on the shared Postgres server. See Week 7 Gotcha #8. - cur.execute(""" - CREATE TABLE IF NOT EXISTS weather_readings ( - id SERIAL PRIMARY KEY, - city TEXT NOT NULL, - temperature REAL NOT NULL, - humidity REAL NOT NULL, - timestamp TEXT NOT NULL - ) - """) - - for _, row in df.iterrows(): + with conn.cursor() as cur: cur.execute( - "INSERT INTO weather_readings (city, temperature, humidity, timestamp) VALUES (%s, %s, %s, %s)", - (row["city"], row["temperature"], row["humidity"], row["timestamp"]), + f"CREATE SCHEMA IF NOT EXISTS {schema}" # noqa: S608 ) + cur.execute(f"SET search_path TO {schema}") # noqa: S608 + + # TODO: Replace 'weather_readings' with a name that describes your data. + cur.execute(""" + CREATE TABLE IF NOT EXISTS weather_readings ( + id SERIAL PRIMARY KEY, + city TEXT NOT NULL, + temperature REAL NOT NULL, + humidity REAL NOT NULL, + timestamp TEXT NOT NULL + ) + """) + + for _, row in df.iterrows(): + cur.execute( + "INSERT INTO weather_readings (city, temperature, humidity, timestamp)" + " VALUES (%s, %s, %s, %s)", + ( + row["city"], + row["temperature"], + row["humidity"], + row["timestamp"], + ), + ) conn.commit() - cur.close() - log.info("Inserted %d rows into Postgres", len(df)) + log.info("Inserted %d rows into %s.weather_readings", len(df), schema) def upload_raw_json(raw_data) -> None: @@ -55,7 +69,9 @@ def upload_raw_json(raw_data) -> None: except ResourceExistsError: pass - blob_name = f"pipeline/{datetime.utcnow().strftime('%Y-%m-%d_%H%M%S')}.json" + blob_name = ( + f"pipeline/{datetime.now(timezone.utc).strftime('%Y-%m-%d_%H%M%S')}.json" + ) container.upload_blob( name=blob_name, data=json.dumps(raw_data).encode("utf-8"),