diff --git a/AI_ASSIST.md b/AI_ASSIST.md index 9b31b94..2083350 100644 --- a/AI_ASSIST.md +++ b/AI_ASSIST.md @@ -10,6 +10,12 @@ TODO: paste your prompt here. +✗ FAIL pipeline.py does not silence the azure SDK logger (logging.getLogger("azure").setLevel(...)) why yet i have this statement + +```python`` +my pipeline runs but my container app job fails + + ## The code or suggestion it returned @@ -17,9 +23,54 @@ TODO: paste your prompt here. ```text TODO: paste the AI output here. ``` +Let us find out why. Run this to see the logs: + +```bash +az containerapp job execution list \ + --name hannahwn-weather-job \ + --resource-group rg-hyf-data \ + --output table +``` + +Get the latest execution name and then check the portal: + +1. Go to **portal.azure.com** +2. Search **hannahwn-weather-job** +3. Click **Execution History** +4. Click the latest failed execution +5. Click **Logs** + +--- + +**Most likely causes since it works locally but fails in Azure:** + +- `SOURCE_NAME` env var missing in the job +- Connection string has quotes or spaces in the job env vars +- Image is outdated — you may have fixed code locally but not rebuilt and pushed the image + +--- + +**Rebuild and push your latest code first:** +```bash +az acr build \ + --registry hyfregistry \ + --image hannahwn-weather-pipeline:latest \ + . +``` + +Then trigger a new run: +```bash +az containerapp job start \ + --name hannahwn-weather-job \ + --resource-group rg-hyf-data +``` +Paste the error from the portal logs and I can tell you exactly what is wrong. ## What I changed after reviewing it +My container still didn't run succesfully + TODO: describe your review here. +pasted more errors to ai till it worked diff --git a/Dockerfile b/Dockerfile index ee3c978..97d5172 100644 --- a/Dockerfile +++ b/Dockerfile @@ -13,9 +13,13 @@ WORKDIR /app # TODO Task 4: copy requirements.txt (must appear before any COPY src command) +COPY requirements.txt . + # TODO Task 4: install dependencies with pip +RUN pip install -r requirements.txt # TODO Task 4: copy the src/ folder +COPY src/ src/ # TODO Task 4: set the CMD to run the pipeline (python -m src.pipeline) -CMD ["python", "-c", "raise SystemExit('Dockerfile not finished: Task 4 still pending')"] +CMD ["python", "-m", "src.pipeline"] diff --git a/README.md b/README.md index fa94600..257c351 100644 --- a/README.md +++ b/README.md @@ -59,6 +59,7 @@ python -m src.pipeline ``` ## Verifying your deployment (Task 5) +## Verification After deploying the Container App Job and triggering a run, capture proof: diff --git a/docs/execution_history.png b/docs/execution_history.png new file mode 100644 index 0000000..8075864 Binary files /dev/null and b/docs/execution_history.png differ diff --git a/requirements.txt b/requirements.txt index b5d18d7..c927e03 100644 --- a/requirements.txt +++ b/requirements.txt @@ -7,7 +7,8 @@ # with explicit pins. # TODO: pin azure-storage-blob (uncomment and add a version) -# azure-storage-blob== +azure-storage-blob==12.19.0 # TODO: pin psycopg2-binary (uncomment and add a version) -# psycopg2-binary== +psycopg2-binary==2.9.9 +python-dotenv==1.0.0 diff --git a/src/pipeline.py b/src/pipeline.py index ef1fc8a..15310ee 100644 --- a/src/pipeline.py +++ b/src/pipeline.py @@ -13,15 +13,25 @@ - Container Job: Data Track/Week 6/week_6__5_container_apps_jobs.md """ +import json import logging import os from datetime import date +from azure.storage.blob import BlobServiceClient +from azure.core.exceptions import ResourceExistsError +import psycopg2 +from contextlib import closing +from dotenv import load_dotenv + logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") logger = logging.getLogger(__name__) # TASK 3 hint: quiet the Azure SDK so its DEBUG output does not drown your own # pipeline logs. The right call lives in Chapter 5 (Viewing logs). +logging.getLogger("azure").setLevel(logging.WARNING) + +load_dotenv() def get_config() -> dict: @@ -31,15 +41,27 @@ def get_config() -> dict: - POSTGRES_URL: full Azure Postgres connection string. - AZURE_STORAGE_CONNECTION_STRING: blob storage account connection string. + Optional: - SOURCE_NAME: logical source label, default "weather". - LOG_LEVEL: not parsed here; the orchestrator sets it via env var. Raise RuntimeError with a clear message if a required variable is missing. """ - raise NotImplementedError( - "Task 3: read POSTGRES_URL and AZURE_STORAGE_CONNECTION_STRING from os.environ" - ) + source_name = os.getenv("SOURCE_NAME", "weather") + POSTGRES_URL = os.getenv("POSTGRES_URL") + AZURE_STORAGE_CONNECTION_STRING = os.getenv("AZURE_STORAGE_CONNECTION_STRING") + + if not POSTGRES_URL: + raise RuntimeError("Environment variable 'POSTGRES_URL' is not set") + if not AZURE_STORAGE_CONNECTION_STRING: + raise RuntimeError("Environment variable 'AZURE_STORAGE_CONNECTION_STRING' is not set") + + return { + "source_name": source_name, + "postgres_url": POSTGRES_URL, + "azure_storage_connection_string": AZURE_STORAGE_CONNECTION_STRING, + } def fetch_records() -> list[dict]: @@ -49,7 +71,29 @@ def fetch_records() -> list[dict]: one dict with a stable key set (for example: station, timestamp, temperature_c, humidity_pct). """ - raise NotImplementedError("Task 3: return a list of at least one record") + logger.info("fetching records from source API") + + return [ + { + "station": "Amsterdam", + "timestamp": "2026-06-09T06:00:00Z", + "temperature_c": 17.4, + "humidity_pct": 72, + }, + { + "station": "Rotterdam", + "timestamp": "2026-06-09T06:00:00Z", + "temperature_c": 16.8, + "humidity_pct": 68, + }, + { + "station": "Almere", + "timestamp": "2026-06-09T06:00:00Z", + "temperature_c": 15.9, + "humidity_pct": 75, + }, + ] + def upload_raw_to_blob(records: list[dict], blob_conn_str: str, source: str) -> str: @@ -62,7 +106,28 @@ def upload_raw_to_blob(records: list[dict], blob_conn_str: str, source: str) -> teacher has pre-created it). Overwrite if the blob already exists so same-day reruns succeed. """ - raise NotImplementedError("Task 1 + Task 3: upload records to blob storage") + logger.info("connecting to blob service") + blob_service_client = BlobServiceClient.from_connection_string(blob_conn_str) + + container_name = "raw" + blob_path = f"{source}/{date.today().isoformat()}.json" + + # Ensure the container exists so uploads don't fail with PathNotFoundError. + try: + logger.info("ensuring container '%s' exists", container_name) + blob_service_client.create_container(container_name) + except ResourceExistsError: + pass + except Exception: + logger.exception("failed to create or access container '%s'", container_name) + raise + + blob_client = blob_service_client.get_blob_client(container=container_name, blob=blob_path) + + # Always upload and overwrite same-day blobs so reruns succeed. + logger.info("uploading blob %s (overwrite=true)", blob_path) + blob_client.upload_blob(data=json.dumps(records), overwrite=True) + return blob_path def write_to_postgres(records: list[dict], postgres_url: str) -> int: @@ -78,7 +143,47 @@ def write_to_postgres(records: list[dict], postgres_url: str) -> int: See Chapter 4 for the connection-and-cursor pattern this is based on. """ - raise NotImplementedError("Task 2 + Task 3: insert rows into Azure Postgres") + + with closing(psycopg2.connect(postgres_url)) as conn: + with conn.cursor() as cur: + cur.execute("DROP TABLE IF EXISTS weather_readings") + cur.execute( + """ + CREATE TABLE IF NOT EXISTS weather_readings ( + station TEXT NOT NULL, + timestamp TIMESTAMPTZ NOT NULL, + temperature_c REAL NOT NULL, + humidity_pct INTEGER NOT NULL, + PRIMARY KEY (station, timestamp) + ) + """ + ) + + insert_sql = """ + INSERT INTO weather_readings ( + station, + timestamp, + temperature_c, + humidity_pct + ) VALUES (%s, %s, %s, %s) + ON CONFLICT (station, timestamp) DO UPDATE SET + temperature_c = EXCLUDED.temperature_c, + humidity_pct = EXCLUDED.humidity_pct + """ + + for record in records: + cur.execute( + insert_sql, + ( + record["station"], + record["timestamp"], + record["temperature_c"], + record["humidity_pct"], + ), + ) + conn.commit() + + return len(records) def run() -> None: