diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..5c0cba7 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,58 @@ +# --- Python Bytecode & Caches --- +__pycache__/ +*.pyc +*.pyo +*.pyd +.pytest_cache/ +.tox/ +.nox/ +.coverage +htmlcov/ +.mypy_cache/ +.ruff_cache/ +.black + +# --- Virtual Environments --- +# (Always let Docker build its own environment via requirements.txt or poetry) +.venv/ +venv/ +env/ +ENV/ +pip-log.txt +pip-delete-this-directory.txt + +# --- Build & Distribution --- +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# --- Jupyter Notebooks & Lab --- +.ipynb_checkpoints +.virtual_documents/ + +# --- Environment Secrets & Configs --- +.env +.env.* +!.env.example +*.pod +*.pem +*.key + +# --- IDEs & OS Junk --- +.vscode/ +.idea/ +.DS_Store +Thumbs.db \ No newline at end of file diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..3618bcc --- /dev/null +++ b/.env.example @@ -0,0 +1,6 @@ +# Copy to .env for local runs: cp .env.example .env +# Do not commit .env (it is gitignored). + +API_KEY=test +OUTPUT_DIR=output +GITHUB_USERNAME= diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 51fcbfd..f3662b4 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1,8 +1,3 @@ -# Task 6: Build a CI workflow that runs on pull requests and pushes to main. -# -# See the assignment chapter for the required steps and commands. -# Fill in the TODO values below. -# # NOTE on Task 7 (ACR push): you do NOT need to add `azure/login` or `docker # push` steps to this file. GitHub blocks repository secrets from running on # pull_request workflows that come from a fork, so an `azure/login` step here @@ -15,7 +10,7 @@ name: CI on: push: - branches: ["TODO-replace-with-main"] + branches: ["main"] pull_request: jobs: @@ -29,10 +24,27 @@ jobs: - name: Install dependencies run: pip install -r requirements.txt - name: Lint - run: echo "TODO implement this step" + run: ruff check src tests - name: Format - run: echo "TODO implement this step" + run: ruff format --check src tests - name: Test - run: echo "TODO implement this step" + run: API_KEY=test pytest -q - name: Build image - run: echo "TODO implement this step" + run: docker build -t noneeeed-pipeline:${{ github.sha }} . + # The next three steps need the AZURE_CREDENTIALS secret, which GitHub + # withholds on pull_request runs from forks for security. The `if:` guard + # skips them on fork PRs so CI stays green; the central `Grade ACR push` + # workflow on main handles the actual push from base-repo context. + - name: Azure login + if: github.event_name != 'pull_request' || github.event.pull_request.head.repo.fork == false + uses: azure/login@v2 + with: + creds: ${{ secrets.AZURE_CREDENTIALS }} + - name: ACR login + if: github.event_name != 'pull_request' || github.event.pull_request.head.repo.fork == false + run: az acr login --name hyfregistry + - name: Push image + if: github.event_name != 'pull_request' || github.event.pull_request.head.repo.fork == false + run: | + docker tag noneeeed-pipeline:${{ github.sha }} hyfregistry.azurecr.io/noneeeed-pipeline:${{ github.sha }} + docker push hyfregistry.azurecr.io/noneeeed-pipeline:${{ github.sha }} \ No newline at end of file diff --git a/.gitignore b/.gitignore index ec8f344..cd28dfb 100644 --- a/.gitignore +++ b/.gitignore @@ -170,3 +170,4 @@ venv/ dist/ build/ output/ +data/ diff --git a/AI_ASSIST.md b/AI_ASSIST.md index 70463b0..7ea60a0 100644 --- a/AI_ASSIST.md +++ b/AI_ASSIST.md @@ -6,7 +6,348 @@ ## The prompt I gave - + TODO: paste your prompt here. @@ -15,11 +356,77 @@ TODO: paste your prompt here. ```python -# TODO: paste the AI-generated code here +# TODO: Your Docker build successfully completed, but now your pipeline code is crashing at runtime because DefaultAzureCredential has absolutely no idea who you are. It cycled through every known Azure authentication method (Environment variables, Managed Identity, Azure CLI, etc.) and they all failed. + +Since you are running this inside a container, passing secrets via Environment Variables is the quickest and cleanest way to get EnvironmentCredential to pick them up. + +Here is exactly how to do it. +Step 1: Get Your Azure Service Principal Secrets + +To authenticate via environment variables, you need a Service Principal. If you don't have one, you can create it via the Azure CLI locally: +Bash + +az ad sp create-for-rbac --name "my-pipeline-sp" --role "Storage Blob Data Reader" --scope "/subscriptions//resourceGroups/" + +This will output a JSON object. You need three specific values from it: + + AZURE_CLIENT_ID (The appId) + + AZURE_CLIENT_SECRET (The password) + + AZURE_TENANT_ID (The tenant) + +Step 2: How to Pass Them to Your Container + +Never hardcode these values directly inside your Dockerfile, otherwise anyone with access to your image can steal your Azure credentials. Instead, inject them at runtime using one of the methods below. +Method A: Using docker run flags (Quickest) + +If you are running the container manually via the command line, use the -e flag to pass them in: +Bash + +docker run -d \ + -e AZURE_CLIENT_ID="your-client-id-here" \ + -e AZURE_CLIENT_SECRET="your-client-secret-here" \ + -e AZURE_TENANT_ID="your-tenant-id-here" \ + your-image-name + +Method B: Using a .env file (Cleanest for local dev) + + Create a file named .env in your project root directory (and add it to your .gitignore!). + + Populate it with the exact variable names DefaultAzureCredential looks for: + Code snippet + + AZURE_CLIENT_ID=your-client-id-here + AZURE_CLIENT_SECRET=your-client-secret-here + AZURE_TENANT_ID=your-tenant-id-here + + Run your container by referencing the file: + Bash + + docker run --env-file .env your-image-name + +Method C: Using Docker Compose + +If you are using a docker-compose.yml file, you can map them under the environment block. It can read straight from your local shell or a .env file automatically: +YAML + +version: '3.8' +services: + pipeline: + image: your-image-name + environment: + - AZURE_CLIENT_ID=${AZURE_CLIENT_ID} + - AZURE_CLIENT_SECRET=${AZURE_CLIENT_SECRET} + - AZURE_TENANT_ID=${AZURE_TENANT_ID} + +Why This Works + +The DefaultAzureCredential() object in your pipeline.py file follows a very strict chain of command. By setting AZURE_CLIENT_ID, AZURE_CLIENT_SECRET, and AZURE_TENANT_ID, the very first mechanism it tries (EnvironmentCredential) will instantly succeed, and it won't even try hitting that failing 169.254.169.254 IMDS endpoint anymore. ``` ## What I changed after reviewing it -TODO: describe your review and any changes you made. +kept on trying solutions for 12 hours using 4 different AI and the result was the same. authorization mismatch diff --git a/Dockerfile b/Dockerfile index d665b11..1077f64 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,24 +1,10 @@ -# Task 4: Write a cache-friendly Dockerfile. -# -# Requirements (in order): -# 1. Use python:3.11-slim as the base image. -# 2. Copy requirements.txt BEFORE copying source code. -# 3. Install dependencies from requirements.txt. -# 4. Copy src/ into the image. -# 5. Set the CMD to run the pipeline: python -m src.pipeline -# -# Replace each TODO comment with the correct Dockerfile instruction. - -# TODO: set the base image -FROM TODO - +FROM python:3.11-slim WORKDIR /app - -# TODO: copy requirements.txt (before source — this keeps the install layer cached) - -# TODO: install dependencies - -# TODO: copy source code - -# TODO: set the command that runs when the container starts -CMD ["TODO"] +COPY requirements.txt . +RUN pip install -r requirements.txt +COPY src/ ./src/ +RUN curl -sL https://aka.ms/InstallAzureCLIDeb | bash +RUN apt-get update && apt-get install -y curl ca-certificates \ +&& curl -sL https://aka.ms/InstallAzureCLIDeb | bash \ +&& rm -rf /var/lib/apt/lists/* +CMD ["python", "-m", "src.pipeline"] \ No newline at end of file diff --git a/assets/acr_push_week5.png b/assets/acr_push_week5.png new file mode 100644 index 0000000..904967a Binary files /dev/null and b/assets/acr_push_week5.png differ diff --git a/requirements.txt b/requirements.txt index 42299cf..c012107 100644 --- a/requirements.txt +++ b/requirements.txt @@ -11,3 +11,12 @@ # ruff== # # Add your pinned dependencies below: +azure-identity==1.25.3 +azure-storage-blob==12.29.0 +pandas==3.0.3 +matplotlib==3.10.9 +pyarrow==24.0.0 +requests==2.34.2 +python-dotenv==1.1.0 +pytest==9.0.3 +ruff==0.15.15 diff --git a/src/clean.py b/src/clean.py new file mode 100644 index 0000000..c4b6822 --- /dev/null +++ b/src/clean.py @@ -0,0 +1,62 @@ +"""Tasks 2 and 3: Explore and clean the raw DataFrames.""" + +from pathlib import Path +import pandas as pd +import logging + +logger = logging.getLogger(__name__) + + +def load_and_explore(data_dir: Path) -> tuple[pd.DataFrame, pd.DataFrame]: + """Task 2: Load both CSV files and explore their contents before cleaning.""" + # TODO: Read messy_sales.csv and messy_customers.csv with pd.read_csv(). + # TODO: For each DataFrame call .info(), .describe(), .head(20), and .isna().sum(). + # TODO: Log what you discover (e.g. which columns have nulls, any suspicious values). + messyS = pd.read_csv(data_dir / "messy_sales.csv") + messyC = pd.read_csv(data_dir / "messy_customers.csv") + logger.info("--- messy sales info ---") + messyS.info() + logger.info("--- describe ---") + logger.info(messyS.describe()) + logger.info("--- head ---") + logger.info(messyS.head(20)) + logger.info("--- missing values ---") + logger.info(messyS.isna().sum()) + logger.info("--- messy customers info ---") + messyC.info() + logger.info("--- describe ---") + logger.info(messyC.describe()) + logger.info("--- head ---") + logger.info(messyC.head(20)) + logger.info("--- missing values ---") + logger.info(messyC.isna().sum()) + return messyS, messyC + + +def clean_sales(sales: pd.DataFrame) -> pd.DataFrame: + """Task 3: Clean the sales DataFrame using vectorized Pandas operations.""" + # TODO: Normalize product_name with .str.strip().str.title(). + sales["product_name"] = sales["product_name"].str.strip().str.title() + # TODO: Normalize customer_email with .str.lower().str.strip(). + sales["customer_email"] = sales["customer_email"].str.lower().str.strip() + # TODO: Convert price to numeric with pd.to_numeric(errors="coerce"). + sales["price"] = pd.to_numeric(sales["price"], errors="coerce") + # TODO: Parse date with pd.to_datetime(errors="coerce"). + sales["date"] = pd.to_datetime(sales["date"], errors="coerce") + # TODO: Drop rows where product_name is missing. + sales = sales.dropna(subset=["product_name"]) + # TODO: Drop rows where price is negative. + sales = sales[sales["price"] >= 0] + # TODO: Drop rows where quantity is zero. + sales = sales[sales["quantity"] != 0] + # TODO: Drop rows where date is NaT (invalid after parsing). + sales = sales.dropna(subset=["date"]) + # TODO: Remove duplicate transactions: .drop_duplicates(subset="transaction_id", keep="first"). + sales = sales.drop_duplicates(subset="transaction_id", keep="first") + # TODO: Decide what to do with outlier prices (clip, flag, or leave) and add a comment explaining why. + product_std = sales.groupby("product_name")["price"].transform("std") + product_mean = sales.groupby("product_name")["price"].transform("mean") + sales["is_price_outlier"] = sales["price"] > (product_mean + 3 * product_std) + # since these are sales transactions, we want to keep outliers (they may be real sales of expensive items) + # but flag them for later just in case for future analysts who may want to filter them out. This way we preserve the data but keep it tracked. + return sales diff --git a/src/ingest.py b/src/ingest.py new file mode 100644 index 0000000..7ab61f7 --- /dev/null +++ b/src/ingest.py @@ -0,0 +1,56 @@ +"""Task 1: Download inputs from Azure. Task 7: Upload outputs back to Azure.""" + +import logging +import os +from pathlib import Path +from azure.identity import DefaultAzureCredential +from azure.storage.blob import BlobServiceClient + +logger = logging.getLogger(__name__) + +ACCOUNT_URL = "https://sthyfstudentsdemo.blob.core.windows.net" +SOURCE_CONTAINER = "week4-inputs" +FILES = ["messy_sales.csv", "messy_customers.csv"] + + +def download_inputs(data_dir: Path) -> None: + """Task 1: Download input CSV files from Azure Blob Storage.""" + logger.info("Initializing Azure credentials...") + # credential = DefaultAzureCredential() + # service = BlobServiceClient(account_url=ACCOUNT_URL, credential=credential) + conn = os.environ.get("AZURE_STORAGE_CONNECTION_STRING") + service = BlobServiceClient.from_connection_string(conn) + container = service.get_container_client(SOURCE_CONTAINER) + + data_dir.mkdir(parents=True, exist_ok=True) + logger.info(f"Target directory verified at: {data_dir.resolve()}") + + for name in FILES: + logger.info(f"Attempting to download {name}...") + blob = container.get_blob_client(name) + + file_path = data_dir / name + with open(file_path, "wb") as f: + f.write(blob.download_blob().readall()) + + logger.info("Downloaded %s to %s", name, file_path) + logger.info(f"Successfully downloaded: {name}") + + +if __name__ == "__main__": + logger.info("Script started...") + target_directory = Path("./data") + download_inputs(target_directory) + logger.info("Script finished.") + + +def upload_outputs(output_dir: Path, github_username: str) -> None: + """Task 7 (extra credit): Upload Parquet outputs to Azure and verify the round-trip.""" + _ = output_dir, github_username # week4-{github_username} when implemented + + # EXTRA CREDIT — implement this after Tasks 2–6 are working. + # TODO: Create a BlobServiceClient using DefaultAzureCredential and ACCOUNT_URL. + # TODO: Get (or create) the container named container_name. + # TODO: Upload every .parquet file in output_dir to the container. + # TODO: Download customer_summary.parquet back and assert its row count matches the local file. + # TODO: Log the container name and number of files uploaded. diff --git a/src/pipeline.py b/src/pipeline.py index 1cd17a4..6e5691b 100644 --- a/src/pipeline.py +++ b/src/pipeline.py @@ -9,22 +9,51 @@ """ import logging +import os from pathlib import Path +from src.ingest import download_inputs, upload_outputs +from src.clean import load_and_explore, clean_sales +from src.transform import join_customers +from src.report import build_reports, write_outputs + +_ROOT = Path(__file__).resolve().parent.parent +_env_file = _ROOT / ".env" +if _env_file.is_file(): + from dotenv import load_dotenv + + load_dotenv(_env_file) + logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") logger = logging.getLogger(__name__) +DATA_DIR = Path("data") +DEFAULT_OUTPUT_DIR = "output" + def get_config() -> dict: """ Return configuration read from environment variables. - Required variable: API_KEY + Required variables: API_KEY, GITHUB_USERNAME Optional variable: OUTPUT_DIR (default "output") Raise RuntimeError with a clear message if a required variable is missing. """ - raise NotImplementedError("Task 5: read API_KEY and OUTPUT_DIR from the environment") + api_key = os.environ.get("API_KEY") + if not api_key: + raise RuntimeError("API_KEY environment variable is required but not set") + github_username = os.environ.get("GITHUB_USERNAME") + if not github_username: + raise RuntimeError( + "GITHUB_USERNAME environment variable is required but not set" + ) + output_dir = os.environ.get("OUTPUT_DIR", DEFAULT_OUTPUT_DIR) + return { + "api_key": api_key, + "output_dir": output_dir, + "github_username": github_username, + } def fetch_data(api_key: str) -> list[dict]: @@ -34,7 +63,8 @@ def fetch_data(api_key: str) -> list[dict]: Return a list of at least one dict representing a record. In a real pipeline you would call requests.get(...) here. """ - raise NotImplementedError("Task 1: return at least one sample record") + _ = api_key # reserved for a real HTTP client + return [{"id": 1, "source": "api", "status": "ok"}] def save_results(records: list[dict], output_dir: Path) -> None: @@ -44,7 +74,12 @@ def save_results(records: list[dict], output_dir: Path) -> None: Create output_dir if it does not exist. Log the number of records written. """ - raise NotImplementedError("Task 1: write records to output_dir/results.txt") + output_dir.mkdir(parents=True, exist_ok=True) + results_file = output_dir / "results.txt" + with results_file.open("w", encoding="utf-8") as f: + for record in records: + f.write(f"{record}\n") + logger.info("wrote %d records to %s", len(records), results_file) def run() -> None: @@ -54,6 +89,19 @@ def run() -> None: output_dir = Path(config["output_dir"]) save_results(records, output_dir) logger.info("pipeline complete") + download_inputs(DATA_DIR) + + sales_raw, customers_raw = load_and_explore(DATA_DIR) + + sales_clean = clean_sales(sales_raw) + enriched = join_customers(sales_clean, customers_raw) + + reports = build_reports(enriched) + write_outputs(reports, output_dir) + + upload_outputs(output_dir, config["github_username"]) + + logging.info("Pipeline complete.") if __name__ == "__main__": diff --git a/src/report.py b/src/report.py new file mode 100644 index 0000000..2813c76 --- /dev/null +++ b/src/report.py @@ -0,0 +1,119 @@ +"""Tasks 5 and 6: Build report tables and write outputs.""" + +import logging +from pathlib import Path + +import matplotlib + +matplotlib.use("Agg") + +import matplotlib.pyplot as plt +import pandas as pd + + +def build_reports(enriched: pd.DataFrame) -> dict[str, pd.DataFrame]: + """Task 5: Build four summary tables using groupby and named aggregations.""" + + # Add ISO week number + enriched["week"] = enriched["date"].dt.isocalendar().week.astype(int) + + # Weekly revenue report + weekly_revenue = ( + enriched.groupby(["week", "region"]) + .agg( + total_revenue=("line_revenue", "sum"), + order_count=("transaction_id", "count"), + ) + .reset_index() + ) + + # Customer summary report + customer_summary = ( + enriched.groupby("customer_email") + .agg( + customer_name=("customer_name", "first"), + region=("region", "first"), + loyalty_tier=("loyalty_tier", "first"), + total_spent=("line_revenue", "sum"), + avg_order=("line_revenue", "mean"), + order_count=("transaction_id", "count"), + ) + .reset_index() + ) + + # Category performance report + category_performance = ( + enriched.groupby("category") + .agg( + total_revenue=("line_revenue", "sum"), + order_count=("transaction_id", "count"), + ) + .reset_index() + ) + + # Loyalty analysis report + loyalty_analysis = ( + enriched.groupby("loyalty_tier") + .agg( + avg_spent=("line_revenue", "mean"), + customer_count=("customer_email", "nunique"), + ) + .reset_index() + ) + + return { + "weekly_revenue": weekly_revenue, + "customer_summary": customer_summary, + "category_performance": category_performance, + "loyalty_analysis": loyalty_analysis, + } + + +def write_outputs( + reports: dict[str, pd.DataFrame], + output_dir: Path, +) -> None: + """Task 6: Write report tables to CSV/Parquet and save a bar chart.""" + + # Create output folder + output_dir.mkdir(parents=True, exist_ok=True) + + # Write report tables + reports["weekly_revenue"].to_csv( + output_dir / "weekly_revenue.csv", + index=False, + ) + + reports["customer_summary"].to_parquet( + output_dir / "customer_summary.parquet", + index=False, + ) + + reports["category_performance"].to_csv( + output_dir / "category_performance.csv", + index=False, + ) + + # Sort category performance table + category_sorted = reports["category_performance"].sort_values( + by="total_revenue", + ascending=False, + ) + + # Create sanity-check chart + category_sorted.plot( + kind="bar", + x="category", + y="total_revenue", + title="Revenue by category", + ) + + # Save chart + plt.savefig( + output_dir / "category_revenue.png", + bbox_inches="tight", + ) + + plt.close() + + logging.info("Reports written to %s", output_dir) diff --git a/src/transform.py b/src/transform.py new file mode 100644 index 0000000..94b02ea --- /dev/null +++ b/src/transform.py @@ -0,0 +1,17 @@ +"""Task 4: Join customer data and add derived columns.""" + +import pandas as pd + + +def join_customers(sales: pd.DataFrame, customers: pd.DataFrame) -> pd.DataFrame: + """Task 4: Normalize join keys, merge, and add a derived boolean flag.""" + # TODO: Normalize customer_email in both DataFrames with .str.lower().str.strip(). + sales["customer_email"] = sales["customer_email"].str.lower().str.strip() + customers["customer_email"] = customers["customer_email"].str.lower().str.strip() + # TODO: Merge sales with customers on customer_email using an inner join. + merged = pd.merge(sales, customers, on="customer_email", how="inner") + # TODO: Add a vectorized boolean column is_high_value: True where price * quantity >= 150. + merged["line_revenue"] = merged["price"] * merged["quantity"] + merged["is_high_value"] = merged["line_revenue"] >= 150 + # TODO: (Optional hands-on) Try a left join instead and inspect rows where customer_name is NaN. + return merged diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 73029e3..868fb45 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -1,25 +1,34 @@ -"""Tests for the Week 5 pipeline.""" +"""Tests for pipeline helper functions (cleaning, transforming, validating).""" +import os + +import pandas as pd import pytest -from src.pipeline import fetch_data, get_config, save_results +from src.clean import clean_sales +from src.pipeline import get_config +from src.transform import join_customers +from src.pipeline import fetch_data, save_results class TestGetConfig: def test_returns_api_key_from_env(self, monkeypatch): monkeypatch.setenv("API_KEY", "test-key-123") + monkeypatch.setenv("GITHUB_USERNAME", "test-user") monkeypatch.delenv("OUTPUT_DIR", raising=False) config = get_config() assert config["api_key"] == "test-key-123" def test_uses_default_output_dir(self, monkeypatch): monkeypatch.setenv("API_KEY", "test-key-123") + monkeypatch.setenv("GITHUB_USERNAME", "test-user") monkeypatch.delenv("OUTPUT_DIR", raising=False) config = get_config() assert config["output_dir"] == "output" def test_reads_custom_output_dir(self, monkeypatch): monkeypatch.setenv("API_KEY", "test-key-123") + monkeypatch.setenv("GITHUB_USERNAME", "test-user") monkeypatch.setenv("OUTPUT_DIR", "/tmp/myout") config = get_config() assert config["output_dir"] == "/tmp/myout" @@ -59,3 +68,47 @@ def test_file_contains_records(self, tmp_path): save_results([{"id": 1}, {"id": 2}], tmp_path) content = (tmp_path / "results.txt").read_text() assert len(content.strip().splitlines()) >= 2 + + +def test_clean_sales_normalizes_product_name(): + sales = pd.DataFrame( + { + "transaction_id": [1], + "product_name": [" widget "], + "customer_email": ["A@X.COM"], + "price": [10.0], + "quantity": [1], + "date": ["2024-01-01"], + } + ) + result = clean_sales(sales) + assert result.iloc[0]["product_name"] == "Widget" + + +def test_join_customers_adds_high_value_flag(): + sales = pd.DataFrame( + { + "customer_email": ["a@b.com"], + "price": [100.0], + "quantity": [2], + } + ) + customers = pd.DataFrame( + { + "customer_email": ["a@b.com"], + "customer_name": ["Alice"], + "region": ["EU"], + "loyalty_tier": ["gold"], + "category": ["tech"], + } + ) + result = join_customers(sales, customers) + assert bool(result.iloc[0]["is_high_value"]) + assert result.iloc[0]["line_revenue"] == 200.0 + + +def test_get_config_raises_when_api_key_missing(): + os.environ["GITHUB_USERNAME"] = "test-user" + os.environ.pop("API_KEY", None) + with pytest.raises(RuntimeError, match="API_KEY"): + get_config()