diff --git a/AI_ASSIST.md b/AI_ASSIST.md index 9b31b94..215a103 100644 --- a/AI_ASSIST.md +++ b/AI_ASSIST.md @@ -6,20 +6,21 @@ ## The prompt I gave - - -TODO: paste your prompt here. +I used ChatGPT to help me understand Azure Blob Storage, Azure PostgreSQL, Docker images, and Azure Container App Jobs. I also asked for help troubleshooting deployment errors, Docker package issues, Azure CLI commands, and job execution failures. ## The code or suggestion it returned - - ```text -TODO: paste the AI output here. +ChatGPT explained Azure concepts, suggested debugging steps, helped interpret error messages, explained missing Python package errors, and provided examples of Azure CLI commands for deployment, verification, and troubleshooting. ``` ## What I changed after reviewing it - +I verified: -TODO: describe your review here. +- the required Python packages were installed correctly; +- the Docker image built successfully; +- the image was pushed to Azure Container Registry; +- the Azure Container App Job was created and executed successfully; +- a JSON file was uploaded to Azure Blob Storage; +- the execution history showed a successful run. \ No newline at end of file diff --git a/Dockerfile b/Dockerfile index ee3c978..4afdb01 100644 --- a/Dockerfile +++ b/Dockerfile @@ -11,11 +11,10 @@ FROM python:3.11-slim 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 --no-cache-dir -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..02c3ff4 100644 --- a/README.md +++ b/README.md @@ -58,18 +58,16 @@ set -a && source .env && set +a python -m src.pipeline ``` -## Verifying your deployment (Task 5) - -After deploying the Container App Job and triggering a run, capture proof: - -1. Open the Azure portal, find your Container App Job, open the **Execution - history** blade. -2. Screenshot the most recent successful run. -3. Save the screenshot to `docs/`. -4. Replace this whole section with one called `## Verification` and embed - your screenshot using a Markdown image link. The grader looks for the - `## Verification` heading and a `![alt](docs/your-file.png)` reference - pointing at the image you committed. +## Verification + +![Execution history](docs/execution_history.png) + +Verification commands: + +```bash +az containerapp job execution list --name halyna1995-job --resource-group rg-hyf-data --output table + +az storage blob list --account-name hyfstoragedev --container-name raw --auth-mode login --output table ## Check your score locally diff --git a/docs/execution_history.png b/docs/execution_history.png new file mode 100644 index 0000000..74017a4 Binary files /dev/null and b/docs/execution_history.png differ diff --git a/requirements.txt b/requirements.txt index b5d18d7..67b456c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,8 +6,8 @@ # The pipeline needs the Azure Blob SDK and a Postgres driver. Add them below # with explicit pins. -# TODO: pin azure-storage-blob (uncomment and add a version) -# azure-storage-blob== +azure-storage-blob==12.14.0 -# TODO: pin psycopg2-binary (uncomment and add a version) -# psycopg2-binary== +psycopg2-binary==2.9.10 + +six==1.16.0 \ No newline at end of file diff --git a/src/pipeline.py b/src/pipeline.py index ef1fc8a..74b663c 100644 --- a/src/pipeline.py +++ b/src/pipeline.py @@ -17,8 +17,22 @@ import os from datetime import date +import json +from contextlib import closing + +try: + import psycopg2 # type: ignore[import] +except ImportError as exc: + raise ImportError( + """psycopg2 is required to run this pipeline. + Install it with `pip install psycopg2-binary`.""" + ) from exc +from azure.storage.blob import BlobServiceClient + + logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") logger = logging.getLogger(__name__) +logging.getLogger("azure").setLevel(logging.WARNING) # 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). @@ -37,9 +51,24 @@ def get_config() -> dict: 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" - ) + + postqres_url = os.environ.get("POSTGRES_URL") + storage_connection = os.environ.get("AZURE_STORAGE_CONNECTION_STRING") + source_name = os.environ.get("SOURCE_NAME", "weather") + + missing = [] + if not postqres_url: + missing.append("POSTGRES_URL") + if not storage_connection: + missing.append("AZURE_STORAGE_CONNECTION_STRING") + if missing: + raise RuntimeError(f"Missing environment variables: {', '.join(missing)}") + + return { + "postgres_url": postqres_url, + "azure_storage_connection": storage_connection, + "source_name": source_name, + } def fetch_records() -> list[dict]: @@ -49,7 +78,14 @@ 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") + return [ + { + "station": "Eindhoven", + "timestamp": f"{date.today().isoformat()}T6:00:00Z", + "temperature_c": 18.5, + "humidity_pct": 72, + } + ] def upload_raw_to_blob(records: list[dict], blob_conn_str: str, source: str) -> str: @@ -62,7 +98,15 @@ 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") + blob_name = f"raw/{source}/{date.today().isoformat()}.json" + client = BlobServiceClient.from_connection_string(blob_conn_str) + container_client = client.get_container_client("raw") + + data = json.dumps(records, indent=2).encode("utf-8") + + container_client.upload_blob(blob_name, data, overwrite=True) + + return blob_name def write_to_postgres(records: list[dict], postgres_url: str) -> int: @@ -78,17 +122,52 @@ 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 cursor: + cursor.execute( + """ + CREATE TABLE IF NOT EXISTS weather ( + station TEXT, + timestamp TIMESTAMPTZ, + temperature_c REAL, + humidity_pct REAL, + PRIMARY KEY (station, timestamp) + ) + """ + ) + + for record in records: + cursor.execute( + """ + INSERT INTO weather ( + 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 + """, + ( + record["station"], + record["timestamp"], + record["temperature_c"], + record["humidity_pct"], + ), + ) + + conn.commit() + return len(records) def run() -> None: + """Run the pipeline end to end.""" config = get_config() logger.info("starting pipeline") records = fetch_records() blob_name = upload_raw_to_blob( records, - config["azure_storage_connection_string"], + config["azure_storage_connection"], config["source_name"], ) logger.info("uploaded blob %s", blob_name)