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..c5bbb59 100644 --- a/README.md +++ b/README.md @@ -58,18 +58,17 @@ 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..37443be 100644 --- a/src/pipeline.py +++ b/src/pipeline.py @@ -15,10 +15,11 @@ import logging import os -from datetime import date -logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") +from sqlalchemy import create_engine, text + 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 +38,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 +65,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,33 +85,63 @@ 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") -def write_to_postgres(records: list[dict], postgres_url: str) -> int: - """Insert (or upsert) records into Azure Postgres. Return the row count. + container_client.upload_blob(blob_name, data, overwrite=True) - 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. + return blob_name - 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 ensure_sslmode_require(postgres_url: str) -> str: + """Ensure Azure PostgreSQL connection uses sslmode=require.""" + if "sslmode=" in postgres_url: + return postgres_url + + separator = "&" if "?" in postgres_url else "?" + return f"{postgres_url}{separator}sslmode=require" + + +def write_to_postgres(df): + """Write weather readings to the personal PostgreSQL schema.""" + postgres_url = os.environ.get("POSTGRES_URL") + + if not postgres_url: + raise ValueError("POSTGRES_URL environment variable is not set.") + + postgres_url = ensure_sslmode_require(postgres_url) + + github_username = os.environ.get("GITHUB_USERNAME", "halyna1995") + schema_name = f"dev_{github_username}" + table_name = "weather_readings" + + engine = create_engine(postgres_url) + + with engine.begin() as connection: + connection.execute(text(f"CREATE SCHEMA IF NOT EXISTS {schema_name}")) + + df.to_sql( + table_name, + engine, + schema=schema_name, + if_exists="replace", + index=False, + ) + + logger.info("Wrote %s rows to %s.%s", len(df), schema_name, table_name) 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)