MohammedA

AI_ASSIST.md

@@ -8,18 +8,31 @@
 
 <!-- Paste the exact prompt you gave an LLM (ChatGPT, Claude, Copilot, etc.). -->
 
-TODO: paste your prompt here.
+help me complete the Azure deployment for my weather pipeline. I am already done doing the pipeline to upload raw JSON to Azure Blob Storage and insert weather rows into Azure Postgres and both the local Python run and Docker run were working after pushing my Docker image to Azure Container Registry I tried to create and verify an Azure Container App Job but Azure CLI kept failing with this error:
+
+ConnectionResetError(10054, 'An existing connection was forcibly closed by the remote host', None, 10054, None)
+
+I
 
 ## The code or suggestion it returned
 
 <!-- Paste the suggestion verbatim — code, shell commands, or both. -->
 
 ```text
-TODO: paste the AI output here.
+ChatGPT explained that the error was not caused by my Python pipeline or Docker image. The pipeline had already worked locally, Docker had worked locally, Blob Storage upload had worked, Postgres insert had worked, and the image tag was visible in ACR. The problem was with Azure CLI on my local Windows/Git Bash environment while running az containerapp commands.
+
+It suggested checking the job list first:
+
+az containerapp job list \
+  --resource-group rg-hyf-data \
+  --output table
+
+When the same Azure CLI connection reset error continued, ChatGPT suggested using Azure Cloud Shell instead of my local terminal, because Cloud Shell runs inside Azure and avoids the local network/CLI connection problem.
 ```
 
 ## What I changed after reviewing it
 
 <!-- Describe what you accepted, rejected, or modified, and why. -->
 
-TODO: describe your review here.
+accepted the suggestion to use Azure Cloud Shell because the problem was not in the pipeline code. Before this, I had already verified that the Python pipeline could upload the blob and write rows to Postgres locally, and I had verified that the Docker image ran correctly.
+In Cloud Shell, I created the Container App Job using my own image

Dockerfile

@@ -12,10 +12,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"]

README.md

@@ -30,14 +30,14 @@ data-assignment-week-6/
 
 ## Where to start
 
-| Step | File | Task in the chapter |
-|---|---|---|
-| 1 | `requirements.txt` | Pin `azure-storage-blob` and `psycopg2-binary` |
-| 2 | `src/pipeline.py` | Implement `get_config`, `upload_raw_to_blob`, `write_to_postgres` (Tasks 1-3) |
-| 3 | `Dockerfile` | Finish the cache-friendly image (Task 4) |
-| 4 | Azure CLI | Deploy as a Container App Job (Task 4-5) |
-| 5 | `docs/execution_history.png` | Add the Execution-history portal screenshot (Task 5) |
-| 6 | `AI_ASSIST.md` | Fill in your AI prompt + review (Task 7) |
+| Step | File                         | Task in the chapter                                                           |
+| ---- | ---------------------------- | ----------------------------------------------------------------------------- |
+| 1    | `requirements.txt`           | Pin `azure-storage-blob` and `psycopg2-binary`                                |
+| 2    | `src/pipeline.py`            | Implement `get_config`, `upload_raw_to_blob`, `write_to_postgres` (Tasks 1-3) |
+| 3    | `Dockerfile`                 | Finish the cache-friendly image (Task 4)                                      |
+| 4    | Azure CLI                    | Deploy as a Container App Job (Task 4-5)                                      |
+| 5    | `docs/execution_history.png` | Add the Execution-history portal screenshot (Task 5)                          |
+| 6    | `AI_ASSIST.md`               | Fill in your AI prompt + review (Task 7)                                      |
 
 ## Open in Codespaces
 
@@ -82,17 +82,17 @@ The grader reports a score out of 100. The passing threshold is 60.
 
 ## Scoring ladder
 
-| Points | What the grader checks |
-|---|---|
-| 10 | Required files exist (Dockerfile, requirements.txt, src/pipeline.py, AI_ASSIST.md, docs/) |
-| 10 | requirements.txt pins `azure-storage-blob` and `psycopg2-binary` |
-| 10 | Dockerfile copies requirements before src (cache-friendly layer order) |
-| 15 | Pipeline reads both env vars, wraps the Postgres connection so it is closed cleanly, and silences the Azure SDK logger |
-| 15 | Pipeline uses an idempotent upsert (`ON CONFLICT ... DO UPDATE`) |
-| 10 | Connection string uses the Azure-required SSL flag and the blob SDK client class |
-| 10 | AI_ASSIST.md has all three sections and is filled in (>=1800 chars, no `TODO:`) |
-| 10 | README has a `## Verification` heading and references an image in `docs/` |
-| 10 | `docs/execution_history.png` exists and is non-trivial (real screenshot) |
+| Points | What the grader checks                                                                                                 |
+| ------ | ---------------------------------------------------------------------------------------------------------------------- |
+| 10     | Required files exist (Dockerfile, requirements.txt, src/pipeline.py, AI_ASSIST.md, docs/)                              |
+| 10     | requirements.txt pins `azure-storage-blob` and `psycopg2-binary`                                                       |
+| 10     | Dockerfile copies requirements before src (cache-friendly layer order)                                                 |
+| 15     | Pipeline reads both env vars, wraps the Postgres connection so it is closed cleanly, and silences the Azure SDK logger |
+| 15     | Pipeline uses an idempotent upsert (`ON CONFLICT ... DO UPDATE`)                                                       |
+| 10     | Connection string uses the Azure-required SSL flag and the blob SDK client class                                       |
+| 10     | AI_ASSIST.md has all three sections and is filled in (>=1800 chars, no `TODO:`)                                        |
+| 10     | README has a `## Verification` heading and references an image in `docs/`                                              |
+| 10     | `docs/execution_history.png` exists and is non-trivial (real screenshot)                                               |
 
 ## Submitting
 
@@ -108,3 +108,17 @@ repo (`Data Track/Week 6/week_6__8_assignment.md`). The auto-grader checks
 code shape, not live Azure deployment, because the GitHub Actions runner has
 no Azure credentials. To rebuild from a fresh scaffold, follow
 `.agents/workflows/build_assignment_repo.md` in the curriculum repo.
+
+## Verification
+
+The Container App Job ran successfully.
+
+Execution history screenshot:
+
+![Execution history](docs/execution_history.png)
+
+CLI output is saved in:
+
+```text
+docs/execution_history.txt
+```

docs/execution_history.txt

@@ -0,0 +1,3 @@
+Name                                     StartTime                  Status
+---------------------------------------  -------------------------  ---------
+mohammedalfakih-dev-weather-job-lq9lmww  2026-06-09T22:20:30+00:00  Succeeded

requirements.txt

@@ -11,3 +11,5 @@
 
 # TODO: pin psycopg2-binary (uncomment and add a version)
 # psycopg2-binary==
+azure-storage-blob==12.24.0
+psycopg2-binary==2.9.10

src/pipeline.py

@@ -13,78 +13,183 @@
 - Container Job:    Data Track/Week 6/week_6__5_container_apps_jobs.md
 """
 
+import json
 import logging
 import os
-from datetime import date
+from contextlib import closing
+from datetime import date, datetime, timezone
+from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
+
+import psycopg2
+from azure.storage.blob import BlobServiceClient
+from psycopg2.extras import execute_values
+
+
+logging.basicConfig(
+    level=os.getenv("LOG_LEVEL", "INFO"),
+    format="%(levelname)s %(message)s",
+)
+
+# Required by the assignment: silence noisy Azure SDK logs.
+logging.getLogger("azure").setLevel(logging.WARNING)
 
-logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
 logger = logging.getLogger(__name__)
 
+SCHEMA_NAME = "dev_mohammedalfakih"
+TABLE_NAME = "weather_readings"
 # 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).
 
+def ensure_sslmode_require(postgres_url: str) -> str:
+    """Ensure Azure Postgres connection string uses sslmode=require."""
+    parts = urlsplit(postgres_url)
+    query = dict(parse_qsl(parts.query))
+
+    if "sslmode" not in query:
+        query["sslmode"] = "require"
+
+    return urlunsplit(
+        (
+            parts.scheme,
+            parts.netloc,
+            parts.path,
+            urlencode(query),
+            parts.fragment,
+        )
+    )
+
 
 def get_config() -> dict:
     """Return configuration read from environment variables.
 
     Required:
-        - POSTGRES_URL: full Azure Postgres connection string.
-        - AZURE_STORAGE_CONNECTION_STRING: blob storage account connection string.
+        POSTGRES_URL
+        AZURE_STORAGE_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.
+        SOURCE_NAME, default "weather"
     """
-    raise NotImplementedError(
-        "Task 3: read POSTGRES_URL and AZURE_STORAGE_CONNECTION_STRING from os.environ"
-    )
+    postgres_url = os.getenv("POSTGRES_URL")
+    blob_conn_str = os.getenv("AZURE_STORAGE_CONNECTION_STRING")
 
+    missing = []
+    if not postgres_url:
+        missing.append("POSTGRES_URL")
+    if not blob_conn_str:
+        missing.append("AZURE_STORAGE_CONNECTION_STRING")
 
-def fetch_records() -> list[dict]:
-    """Return a small batch of records to ingest.
+    if missing:
+        raise RuntimeError(
+            f"Missing required environment variables: {', '.join(missing)}"
+        )
 
-    In a real pipeline you would call an API here. Return a list of at least
-    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 {
+        "postgres_url": ensure_sslmode_require(postgres_url),
+        "azure_storage_connection_string": blob_conn_str,
+        "source_name": os.getenv("SOURCE_NAME", "weather"),
+    }
+
+
+def fetch_records() -> list[dict]:
+    """Return a small batch of weather records to ingest."""
+    now = datetime.now(timezone.utc).replace(microsecond=0).isoformat()
+
+    return [
+        {
+            "station": "Open-Meteo Copenhagen",
+            "timestamp": now,
+            "temperature_c": 12.5,
+            "humidity_pct": 80,
+        },
+        {
+            "station": "Open-Meteo Amsterdam",
+            "timestamp": now,
+            "temperature_c": 10.8,
+            "humidity_pct": 75,
+        },
+    ]
 
 
 def upload_raw_to_blob(records: list[dict], blob_conn_str: str, source: str) -> str:
-    """Upload the raw records as a single JSON blob and return its name.
+    """Upload raw records as JSON to Azure Blob Storage."""
+    blob_name = f"raw/{source}/{date.today().isoformat()}.json"
 
-    The blob name must follow the assignment convention:
-        raw/<source>/<YYYY-MM-DD>.json
+    service = BlobServiceClient.from_connection_string(blob_conn_str)
+    blob_client = service.get_blob_client(
+        container="raw",
+        blob=blob_name,
+    )
 
-    Use the azure-storage-blob SDK. The target container is "raw" (your
-    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")
+    payload = json.dumps(records, indent=2)
 
+    blob_client.upload_blob(
+        payload,
+        overwrite=True,
+        content_type="application/json",
+    )
 
-def write_to_postgres(records: list[dict], postgres_url: str) -> int:
-    """Insert (or upsert) records into Azure Postgres. Return the row count.
+    return blob_name
 
-    Steps:
-        1. Open a psycopg2 connection wrapped so it is closed cleanly when the
-           function returns, even on error.
-        2. Ensure the target table exists (create it if missing).
-        3. Insert each record. The pipeline must be safe to rerun on the same
-           day: a re-run must update rather than fail.
-        4. Commit and return the number of rows written.
 
-    See Chapter 4 for the connection-and-cursor pattern this is based on.
-    """
-    raise NotImplementedError("Task 2 + Task 3: insert rows into Azure Postgres")
+
+def write_to_postgres(records: list[dict], postgres_url: str) -> int:
+    """Upsert weather readings into Azure Postgres and return row count."""
+    with closing(psycopg2.connect(postgres_url)) as conn:
+        with conn:
+            with conn.cursor() as cur:
+                cur.execute(f"CREATE SCHEMA IF NOT EXISTS {SCHEMA_NAME}")
+                cur.execute(f"SET search_path TO {SCHEMA_NAME}")
+
+                cur.execute(
+                    f"""
+                    CREATE TABLE IF NOT EXISTS {TABLE_NAME} (
+                        station TEXT NOT NULL,
+                        timestamp TIMESTAMPTZ NOT NULL,
+                        temperature_c DOUBLE PRECISION NOT NULL,
+                        humidity_pct INTEGER NOT NULL,
+                        ingested_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+                        PRIMARY KEY (station, timestamp)
+                    )
+                    """
+                )
+
+                rows = [
+                    (
+                        record["station"],
+                        record["timestamp"],
+                        record["temperature_c"],
+                        record["humidity_pct"],
+                    )
+                    for record in records
+                ]
+
+                execute_values(
+                    cur,
+                    f"""
+                    INSERT INTO {TABLE_NAME} (
+                        station,
+                        timestamp,
+                        temperature_c,
+                        humidity_pct
+                    )
+                    VALUES %s
+                    ON CONFLICT (station, timestamp)
+                    DO UPDATE SET
+                        temperature_c = EXCLUDED.temperature_c,
+                        humidity_pct = EXCLUDED.humidity_pct,
+                        ingested_at = NOW()
+                    """,
+                    rows,
+                )
+
+    return len(records)
 
 
 def run() -> None:
     config = get_config()
     logger.info("starting pipeline")
     records = fetch_records()
+    logger.info("fetched %d records", len(records))
 
     blob_name = upload_raw_to_blob(
         records,