Hannah

.github/workflows/ci.yml

@@ -15,7 +15,7 @@ name: CI
 
 on:
   push:
-    branches: ["TODO-replace-with-main"]
+    branches: ["main"]
   pull_request:
 
 jobs:
@@ -29,10 +29,20 @@ jobs:
       - name: Install dependencies
         run: pip install -r requirements.txt
       - name: Lint
-        run: echo "TODO implement this step"
+        run: ruff check src
       - name: Format
-        run: echo "TODO implement this step"
+        run: ruff format --check src
       - name: Test
-        run: echo "TODO implement this step"
+        run: pytest -q
       - name: Build image
-        run: echo "TODO implement this step"
+        run: docker build -t hannahwn-pipeline:${{ github.sha }} .
+      - name: Azure login
+        uses: azure/login@v2
+        with:
+             creds: ${{ secrets.AZURE_CREDENTIALS }}
+      - name: ACR login
+        run: az acr login --name hyfregistry
+      - name: Push image
+        run: |
+             docker tag hannahwn-pipeline:${{ github.sha }} hyfregistry.azurecr.io/hannahwn-pipeline:${{ github.sha }}
+             docker push hyfregistry.azurecr.io/hannahwn-pipeline:${{ github.sha }}

AI_ASSIST.md

@@ -10,9 +10,22 @@
 
 TODO: paste your prompt here.
 
+how to solve  mitigate this issue, please refer to the troubleshooting guidelines here at https://aka.ms/azsdk/python/identity/defaultazurecredential/troubleshoot.
+=================================================================== warnings summary ===================================================================
+tests/test_pipeline.py: 12 warnings
+  C:\Users\Gebruiker\c55-data-week-5\venv\Lib\site-packages\msal\token_cache.py:293: DeprecationWarning: Use list(search(...)) instead to explicitly geta list.
+    warnings.warn(
+
+-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
+=============================================================== short test summary info ================================================================
+FAILED tests/test_pipeline.py::TestDownloadInputs::test_downloads_files - azure.core.exceptions.ClientAuthenticationError: DefaultAzureCredential failedto retrieve a token from the included credentials.
+================================================= 1 failed, 10 passed, 12 warnings in 95.82s (0:01:35) =================================================
+(venv) 
+
 ## The code or suggestion it returned
 
 <!-- Paste the code or key suggestion the LLM returned. -->
+the same Azure auth issue.Still the same Azure auth issue. Your token has expired and needs refreshing. Run this:
 
 ```python
 # TODO: paste the AI-generated code here
@@ -23,3 +36,5 @@ TODO: paste your prompt here.
 <!-- Describe what you accepted, rejected, or modified, and why. -->
 
 TODO: describe your review and any changes you made.
+
+There is no change , i still experience problems with loggin in in azure

Dockerfile

@@ -10,15 +10,18 @@
 # 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)
+COPY requirements.txt .
 
 # TODO: install dependencies
+RUN pip install --no-cache-dir -r requirements.txt
 
 # TODO: copy source code
+COPY src/ ./src/
 
 # TODO: set the command that runs when the container starts
-CMD ["TODO"]
+CMD ["python", "-m", "src.pipeline"]

data/messy_customers.csv

@@ -0,0 +1,38 @@
+customer_email,customer_name,region,signup_date,loyalty_tier
+alice@example.com,Alice van den Berg,NL,2023-11-15,Gold
+Bob@Company.COM,Bob De Smet,BE,2023-12-01,Silver
+charlie@work.org,Charlie Müller,DE,2024-01-10,Bronze
+dave@email.com,,NL,2024-01-15,Silver
+eve@startup.io,Eve Jansen,NL,2024-02-01,Gold
+frank@corp.com,Frank Dubois,FR,2024-02-14,Bronze
+Grace@University.EDU,Grace van Dijk,NL,2024-03-01,Silver
+henry@business.com,Henry Peeters,BE,2024-03-10,Bronze
+ivan@email.com,Ivan Schneider,DE,2024-03-15,Silver
+jenny@work.org,Jenny Laurent,FR,2024-01-20,Gold
+karl@startup.io,Karl Bakker,NL,2024-02-28,Bronze
+lena@mail.nl,Lena de Vries,NL,2024-04-01,Silver
+Marco@Business.DE,Marco Weber,DE,2024-04-10,Gold
+nina@university.edu,,NL,2024-04-15,Bronze
+oliver@corp.com,Oliver Martin,FR,2024-05-01,Silver
+Paula@Startup.IO,Paula Visser,NL,2024-05-10,Bronze
+quinn@work.org,Quinn Claes,BE,2024-05-15,Silver
+rachel@email.com,Rachel Schmitt,DE,not_a_date,Gold
+simon@company.com,Simon Leroy,FR,2024-06-01,Bronze
+tina@mail.nl,Tina Meijer,NL,2024-06-05,Silver
+uwe@business.de,Uwe Fischer,DE,2024-06-10,Gold
+vera@mail.nl,Vera Willems,BE,2024-06-15,Bronze
+wendy@startup.io,Wendy van Leeuwen,NL,2024-01-05,Silver
+xander@corp.com,Xander Moreau,FR,2024-02-20,Gold
+Yara@University.EDU,Yara Hendriks,NL,2024-03-25,Bronze
+zach@email.com,Zach Bauer,DE,2024-04-05,Silver
+orphan_customer1@mail.nl,Dirk Janssen,NL,2024-01-01,Gold
+orphan_customer2@business.de,Petra Hofmann,DE,2024-02-15,Bronze
+orphan_customer3@work.fr,Louis Petit,FR,2024-03-20,Silver
+orphan_customer4@mail.be,,BE,invalid_date,Gold
+sam@company.com,Sam de Groot,NL,2024-05-20,Silver
+lisa@email.com,Lisa Maes,BE,2024-06-01,Bronze
+tom@corp.com,Tom Bernard,FR,2024-04-18,Gold
+emma@startup.io,Emma van Houten,NL,2024-03-12,Silver
+jan@mail.nl,Jan de Boer,NL,2024-02-08,Bronze
+sophie@business.de,Sophie Klein,DE,2024-05-25,Gold
+orphan_customer5@university.edu,Maria Garcia,FR,2024-06-20,Silver

requirements.txt

@@ -11,3 +11,11 @@
 # ruff==
 #
 # Add your pinned dependencies below:
+
+pandas==2.2.2
+azure-identity==1.15.0
+azure-storage-blob==12.14.1
+python-dotenv==1.0.0    
+ruff==0.0.241
+pytest==7.3.1   
+

src/clean.py

@@ -0,0 +1,51 @@
+"""Tasks 2 and 3: Explore and clean the raw DataFrames."""
+import logging
+from pathlib import Path
+
+import pandas as pd
+
+
+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().
+    sales_df = pd.read_csv(data_dir / "messy_sales.csv")
+    customers_df = pd.read_csv(data_dir / "messy_customers.csv")
+    # TODO: For each DataFrame call .info(), .describe(), .head(20), and .isna().sum().
+    logging.info("Sales DataFrame info:\n%s", sales_df.info())
+    logging.info("Sales DataFrame description:\n%s", sales_df.describe())
+    logging.info("Sales DataFrame head:\n%s", sales_df.head(20))
+    logging.info("Sales DataFrame null counts:\n%s", sales_df.isna().sum())
+    logging.info("Customers DataFrame info:\n%s", customers_df.info())
+    logging.info("Customers DataFrame description:\n%s", customers_df.describe())
+    logging.info("Customers DataFrame head:\n%s", customers_df.head(20))
+    logging.info("Customers DataFrame null counts:\n%s", customers_df.isna().sum())
+    # TODO: Log what you discover (e.g. which columns have nulls, any suspicious values).
+    logging.info("Sales DataFrame has %d rows and %d columns", sales_df.shape[0], sales_df.shape[1])
+    logging.info("Customers DataFrame has %d rows and %d columns", customers_df.shape[0], customers_df.shape[1])
+    return sales_df, customers_df
+
+
+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.
+    # For simplicity, we'll leave outlier prices as they are, but in a real scenario, we might want to investigate them further or apply business rules to handle them.
+    return sales
+

src/ingest.py

@@ -0,0 +1,69 @@
+"""Task 1: Download inputs from Azure. Task 7: Upload outputs back to Azure."""
+import io
+import logging
+import os
+from dotenv import load_dotenv
+from pathlib import Path
+
+import pandas as pd
+from azure.identity import DefaultAzureCredential
+from azure.storage.blob import BlobServiceClient
+
+load_dotenv()
+
+ACCOUNT_URL = os.getenv("ACCOUNT_URL", "https://c55data.blob.core.windows.net")
+SOURCE_CONTAINER = os.getenv("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."""
+    # TODO: Create a BlobServiceClient using DefaultAzureCredential and ACCOUNT_URL.
+    credential = DefaultAzureCredential()
+    blob_service_client = BlobServiceClient(account_url=ACCOUNT_URL, credential=credential) 
+    # TODO: Get a container client for SOURCE_CONTAINER.
+    container_client = blob_service_client.get_container_client(SOURCE_CONTAINER)
+    # TODO: For each filename in FILES, download the blob and write it to data_dir/<filename>.
+    data_dir.mkdir(parents=True, exist_ok=True)
+
+    for name in FILES:
+        blob = container_client.get_blob_client(name)
+        with open(f"{data_dir}/{name}", "wb") as f:
+             f.write(blob.download_blob().readall())
+
+    # TODO: Log a message for each downloaded file.
+    for filename in FILES:
+        logging.info("Downloaded %s", filename)
+
+
+def upload_outputs(output_dir: Path, github_username: str) -> None:
+    """Task 7 (extra credit): Upload Parquet outputs to Azure and verify the round-trip."""
+    container_name = f"week4-{github_username}"
+
+    # 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.
+    account_url = ACCOUNT_URL
+    credential = DefaultAzureCredential()
+    blob_service_client = BlobServiceClient(account_url=account_url, credential=credential)
+    container_client = blob_service_client.get_container_client(container_name)
+    if not container_client.exists():
+        container_client.create_container()
+    parquet_files = list(output_dir.glob("*.parquet"))
+    for parquet_file in parquet_files:
+        blob_client = container_client.get_blob_client(parquet_file.name)
+        with open(parquet_file, "rb") as f:
+            blob_client.upload_blob(f, overwrite=True)
+    logging.info("Uploaded %d files to container %s", len(parquet_files), container_name)
+    # Verify round-trip for customer_summary.parquet
+    local_file = output_dir / "customer_summary.parquet"
+    blob_client = container_client.get_blob_client(local_file.name)
+    downloaded_data = blob_client.download_blob().readall()
+    downloaded_df = pd.read_parquet(io.BytesIO(downloaded_data))
+    local_df = pd.read_parquet(local_file)
+    assert len(downloaded_df) == len(local_df), "Row count mismatch after round-trip"
+    logging.info("Round-trip verification successful for customer_summary.parquet") 
+

src/pipeline.py

@@ -9,7 +9,14 @@
 """
 
 import logging
+import os
 from pathlib import Path
+DATA_DIR = Path("data")
+
+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
 
 logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
 logger = logging.getLogger(__name__)
@@ -24,7 +31,12 @@ def get_config() -> dict:
 
     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.getenv("API_KEY")
+    if api_key is None:
+        raise RuntimeError("Missing required environment variable: API_KEY")
+    output_dir = os.getenv("OUTPUT_DIR", "output")
+    return {"api_key": api_key, "output_dir": output_dir}
+
 
 
 def fetch_data(api_key: str) -> list[dict]:
@@ -34,7 +46,17 @@ 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")
+    mock_record = {
+        "transaction_id": "12345",
+        "customer_email": "h@gmail.com",
+        "date": "2024-01-01",
+        "region": "North",
+        "category": "Widgets",
+        "quantity": 10,
+        "price": 9.99,
+    }
+    return [mock_record]
+
 
 
 def save_results(records: list[dict], output_dir: Path) -> None:
@@ -44,7 +66,11 @@ 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)
+    with open(output_dir / "results.txt", "w") as f:
+        for record in records:
+            f.write(f"{record}\n")
+    logger.info(f"Written {len(records)} records to {output_dir}/results.txt")
 
 
 def run() -> None:
@@ -55,6 +81,22 @@ def run() -> None:
     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, GITHUB_USERNAME)
+
+    logging.info("Pipeline complete.")
+
 
 if __name__ == "__main__":
     run()
+
+
+