Halyna R.

task-1/output/clean_sales.csv

@@ -0,0 +1,13 @@
+transaction_id,product_name,category,price,quantity,customer_email,date,revenue,vat
+1,Laptop Pro,Electronics,999.99,2,alice@example.com,2024-03-15,1999.98,420.0
+2,Wireless Mouse,Electronics,29.99,5,bob@company.com,2024-03-15,149.95,31.49
+3,Usb Cable,Electronics,4.99,10,,2024-03-16,49.9,10.48
+4,Office Chair,Furniture,349.5,1,charlie@work.org,2024-03-16,349.5,73.39
+5,Standing Desk,Furniture,599.0,1,charlie@work.org,not_a_date,599.0,125.79
+9,Webcam Hd,Electronics,54.99,1,,2024-03-18,54.99,11.55
+10,Desk Lamp,Furniture,34.99,4,grace@university.edu,2024-03-19,139.96,29.39
+11,Noise Cancelling Headphones,Electronics,199.99,1,alice@example.com,2024-03-19,199.99,42.0
+12,Cable Management Kit,Furniture,15.99,6,henry@business.com,2024-03-20,95.94,20.15
+13,Ergonomic Mouse Pad,Furniture,24.99,3,ivan@email.com,2024-03-20,74.97,15.74
+14,Laptop Stand,Furniture,45.99,2,jenny@work.org,2024-03-21,91.98,19.32
+15,Bluetooth Speaker,Unknown,39.99,1,karl@startup.io,2024-03-21,39.99,8.4

task-1/src/config.py

@@ -21,15 +21,11 @@
 
 def _required(name: str) -> str:
     """Read an env var; fail loudly if missing."""
-    # TODO 2: Read os.environ[name]; if not set, raise ValueError with a
-    # message that names the missing variable AND points at .env.example.
-    raise NotImplementedError("Implement _required: see TODO 2 in config.py")
-
-
-# TODO 3: Replace the placeholder lines below by calling _required(...) for
-# each variable. INPUT_PATH and OUTPUT_PATH must be importable from this
-# module by the rest of the pipeline as a relative import
-# (`from .config import INPUT_PATH, ...`), since the pipeline runs as
-# `python -m src.pipeline`.
-INPUT_PATH: str = ""   # TODO: _required("INPUT_PATH")
-OUTPUT_PATH: str = ""  # TODO: _required("OUTPUT_PATH")
+
+    value = os.environ.get(name)
+    if not value:
+        raise ValueError(f"Environment variable '{name}' is not set. Please check .env.example.")
+    return value
+
+INPUT_PATH: str = _required("INPUT_PATH")
+OUTPUT_PATH: str = _required("OUTPUT_PATH")

task-1/src/models.py

@@ -11,23 +11,23 @@
 from dataclasses import dataclass
 
 
-# TODO 1: Define a @dataclass called Transaction with these fields:
-#   transaction_id: int
-#   product_name: str
-#   category: str
-#   price: float
-#   quantity: int
-#   customer_email: str
-#   date: str
-#   revenue: float = 0.0
-#   vat: float = 0.0
-#
-# TODO 2: Add __post_init__ that raises ValueError when:
-#   - self.price < 0  (with a message naming the bad value)
-#   - not self.product_name.strip()  (empty / whitespace-only product name)
+def __post_init__(self):
+    if self.price < 0:
+        raise ValueError(f"Price cannot be negative: {self.price}")
+    if not self.product_name.strip():
+        raise ValueError("Product name cannot be empty or whitespace-only.")
 
-
-# Replace this stub with your dataclass:
 @dataclass
 class Transaction:
-    transaction_id: int  # TODO: replace this stub with the full field list above
+    """A sales transaction."""
+
+    transaction_id: int  
+    product_name: str
+    category: str
+    price: float
+    quantity: int
+    customer_email: str
+    date: str
+    revenue: float = 0.0
+    vat: float = 0.0
+

task-1/src/pipeline.py

@@ -31,29 +31,43 @@
 
 def read_csv(path: str) -> list[dict]:
     """Read a CSV file into a list of dicts. I/O only — no business rules."""
-    # TODO: implement using csv.DictReader.
-    raise NotImplementedError
+    with open(path, newline='', encoding='utf-8') as file:
+        reader = csv.DictReader(file)
+        return list(reader)
 
 
 def write_csv(rows: list[dict], path: str) -> None:
     """Write a list of dicts to CSV. I/O only — no business rules."""
-    # TODO: implement using csv.DictWriter.
-    raise NotImplementedError
-
+    if not rows:
+        return
+
+    with open(path, 'w', newline='', encoding='utf-8') as file:
+        writer = csv.DictWriter(file, fieldnames=rows[0].keys())
+        writer.writeheader()
+        writer.writerows(rows)
 
 def run() -> None:
+    """Run the full pipeline: read, transform, write, summarise."""
     raw = read_csv(INPUT_PATH)
     data = remove_invalid(raw)
     data = clean_fields(data)
     data = filter_zero_quantity(data)
     data = calculate_revenue(data)
 
-    # Materialise as Transaction instances so the dataclass __post_init__
-    # acts as a final guard before serialisation.
-    # TODO: cast price / quantity / revenue / vat to the right types here
-    # if your transforms left them as strings, then iterate over `data`
-    # to build Transaction(**row) for each cleaned row.
-    transactions = [Transaction(**row) for row in data]
+    transactions = [
+        Transaction(
+            transaction_id=int(row['transaction_id']),
+            product_name=row['product_name'],
+            category=row['category'],
+            price=float(row['price']),
+            quantity=int(row['quantity']),
+            customer_email=row['customer_email'],
+            date=row['date'],
+            revenue=float(row['revenue']),
+            vat=float(row['vat'])
+            )
+        for row in data
+    ]
 
     # Output dir must exist. Use pathlib for cross-platform safety.
     Path(OUTPUT_PATH).parent.mkdir(parents=True, exist_ok=True)

task-1/src/transforms.py

@@ -18,9 +18,18 @@ def remove_invalid(rows: list[dict]) -> list[dict]:
 
     Empty here means missing, "", or whitespace-only.
     """
-    # TODO: implement. Return a new list, do not mutate `rows`.
-    raise NotImplementedError
+
+    result = []
+    for row in rows:
+        product_name = str(row.get("product_name") or "").strip()
+        price = float(row.get("price", 0))
 
+        if product_name == "" or price < 0:
+            continue
+
+        result.append({row})
+
+    return result
 
 def clean_fields(rows: list[dict]) -> list[dict]:
     """Clean string fields:
@@ -31,20 +40,59 @@ def clean_fields(rows: list[dict]) -> list[dict]:
 
     Return a new list. Do not mutate the input rows.
     """
-    # TODO: implement.
-    raise NotImplementedError
+
+    result = []
+    for row in rows:
+        product_name = str(row.get("product_name") or "").strip().title()
+        customer_email = str(row.get("customer_email") or "").strip().lower()
+        raw_category = str(row.get("category") or "").strip()
+        category = raw_category if raw_category else "Unknown"
 
+        result.append(
+            {
+                **row,
+                "product_name": product_name,
+                "customer_email": customer_email,
+                "category": category,
+            }
+        )
+
+    return result
 
 def calculate_revenue(rows: list[dict], vat_rate: float = 0.21) -> list[dict]:
     """Add 'revenue' (price * quantity) and 'vat' (revenue * vat_rate) fields.
 
     Round both to 2 decimal places. Coerce price/quantity from string if needed.
     """
-    # TODO: implement.
-    raise NotImplementedError
+
+    result = []
+    for row in rows:
+        price = float(row["price"])
+        quantity = int(row["quantity"])
+        revenue = round(price * quantity, 2)
+        vat = round(revenue * vat_rate, 2)
+
+        result.append(
+            {
+                **row,
+                "price": price,
+                "quantity": quantity,
+                "revenue": revenue,
+                "vat": vat,
+            }
+        )
+
+    return result
 
 
 def filter_zero_quantity(rows: list[dict]) -> list[dict]:
     """Remove rows where quantity is 0."""
-    # TODO: implement.
-    raise NotImplementedError
+
+    result = []
+    for row in rows:
+        if int(row.get("quantity", 0)) == 0:
+            continue
+
+        result.append({**row})
+
+    return result

task-1/tests/test_transforms.py

@@ -6,7 +6,7 @@
   - test_calculate_revenue_adds_fields
   - test_no_mutation
 """
-import pytest
+from copy import deepcopy
 
 from src.transforms import (
     calculate_revenue,
@@ -15,26 +15,41 @@
     remove_invalid,
 )
 
-
-def test_remove_invalid_drops_empty_names():
-    # TODO: feed in 3 rows (one valid, one empty product_name, one
-    # whitespace-only product_name); assert only the valid one survives.
-    raise NotImplementedError
-
-
-def test_clean_fields_normalizes_names():
-    # TODO: feed a row with messy product_name and uppercase email; assert
-    # the output has stripped + title-cased name and lowercase email.
-    raise NotImplementedError
-
-
-def test_calculate_revenue_adds_fields():
-    # TODO: feed a row with price=100, quantity=3; assert output has
-    # revenue=300.0 and vat=63.0 (default VAT rate is 0.21).
-    raise NotImplementedError
-
-
+
 def test_no_mutation():
-    # TODO: feed in a list, run any transform on it, assert the original
-    # list is unchanged. This is the most important test in the file.
-    raise NotImplementedError
+    """Run all transforms and assert the original list is unchanged."""
+    original = [
+        {
+            "transaction_id": "1",
+            "product_name": "  laptop  ",
+            "category": "",
+            "price": "100",
+            "quantity": "2",
+            "customer_email": "  Halyna@gmail.COM  ",
+            "date": "2025-01-15",
+        },
+        {
+            "transaction_id": "2",
+            "product_name": "Mouse",
+            "category": "Accessories",
+            "price": "25",
+            "quantity": "0",
+            "customer_email": "test@example.com",
+            "date": "2025-01-16",
+        },
+    ]
+
+    transforms = [
+        remove_invalid,
+        clean_fields,
+        filter_zero_quantity,
+        calculate_revenue,
+    ]
+
+    for transform in transforms:
+        rows = deepcopy(original)
+        expected = deepcopy(rows)
+
+        transform(rows)
+
+        assert rows == expected