From 80124fce7e689b7a1122ee9bccf7aa98244d4d5d Mon Sep 17 00:00:00 2001 From: Mareh Date: Wed, 20 May 2026 23:49:02 +0200 Subject: [PATCH 1/2] Complete Week3 --- AI_DEBUG.md | 91 +++++++++++++++++++++++++++++++++++++++-- database.py | 53 ++++++++++++++++++++---- ingest_api.py | 52 +++++++++++++++++++---- ingest_files.py | 22 +++++++++- models.py | 6 ++- output/azure_compare.md | 6 +-- pipeline.py | 38 +++++++++++------ test_azure.py | 29 +++++++++++++ validate.py | 17 +++++++- 9 files changed, 274 insertions(+), 40 deletions(-) create mode 100644 test_azure.py diff --git a/AI_DEBUG.md b/AI_DEBUG.md index edf34a5..88e9eee 100644 --- a/AI_DEBUG.md +++ b/AI_DEBUG.md @@ -7,20 +7,103 @@ Document the debugging session below. If everything worked first try, introduce ## The Error - + + ## The Prompt + Include: the error, the relevant code snippet, and what you asked the AI to help with. + $ python ingest_files.py +C:\Users\Beheerder\c55-data-week-3\ingest_files.py:47: SyntaxWarning: invalid escape sequence '\w' +  csv_path = Path("data\weather_stations.csv") +--- STARTING CSV PIPELINE TEST --- +I need to Know the coulmn name of the data: ['station', 'timestamp', 'temperature_c', 'humidity_pct'] + +--- TEST RESULTS --- +Total rows processed from CSV: 10 + +First 3 sample rows look like this: +{'station': 'Copenhagen', 'timestamp': '2025-01-15T10:00', 'temperature_c': 18.5, 'humidity_pct': None} +{'station': '', 'timestamp': '2025-01-15T11:00', 'temperature_c': 20.1, 'humidity_pct': None} +{'station': 'Aarhus', 'timestamp': '2025-01-15T12:00', 'temperature_c': 'N/A', 'humidity_pct': None} + +Last row (to check for messy data parsing): +{'station': 'Kolding', 'timestamp': '2025-01-15T18:00', 'temperature_c': 14.9, 'humidity_pct': None} +i just wanted to understand this file but way i got None for temp and humidity? + --> ## The Solution + Did you apply the suggestion as-is, or did you need to adapt it? Explain what changed. + The output you are seeing shows that your file reading logic is working, but it highlights exactly why data pipelines need a validation step! You are getting `None` and strings like `'N/A'` because the raw CSV file contains messy, missing, or corrupted data. + +Let's break down exactly why `temperature_c` and `humidity_pct` look like that in your sample rows: + +--- + +## 1. Why `humidity_pct` is always `None` + +In your first three sample rows and your last row, **every single `humidity_pct` value is `None`.** There are two likely reasons for this: + +* **The CSV column is completely empty:** The actual `weather_stations.csv` file might have headers for humidity, but the rows below it are blank or missing data entirely for that column. +* **A column name mismatch:** Look closely at your print statement: +```text +I need to Know the coulmn name of the data: ['station', 'timestamp', 'temperature_c', 'humidity_pct'] + +``` + + +If the CSV file spells it slightly differently (for example, `humidity` or `humidity_%`), your Python CSV reader won't find an exact match for `"humidity_pct"`, so it defaults to filling that key with `None`. + +--- + +## 2. Why `temperature_c` has `'N/A'` and missing values + +Look at your third sample row: + +```python +{'station': 'Aarhus', 'timestamp': '2025-01-15T12:00', 'temperature_c': 'N/A', 'humidity_pct': None} + +``` + +The temperature here is the **text string `'N/A'**` instead of a number like `18.5`. This happens because whoever recorded the data at the Aarhus station didn't provide a reading, and typed "N/A" into the file. + +--- + +## This is exactly why Task 4 (Pydantic Validation) exists! + +This test print perfectly proves why your pipeline cannot just blindly push raw CSV data straight into the clean `weather_readings` database table: + +1. **Database Crash Prevention:** Your database table requires `temperature_c` to be a `REAL` (decimal) number and `humidity_pct` to be an `INTEGER`. If you tried to save the text string `'N/A'` or the value `None` into a column that expects numbers, SQLite would throw an error or save broken data. +2. **How Pydantic saves the day:** When these rows hit your `validate_records()` function, Pydantic will flag them: +* The row with `station: ''` (empty string) will fail because station is marked as `NOT NULL` or requires a minimum length. +* The Aarhus row will fail because `'N/A'` cannot be converted into a float number. +* These broken rows will be safely filtered out and sent to your `error_report.json`, leaving only the 100% clean, valid rows to enter your final database! + --> ## Reflection + What would you do differently next time you hit this type of error? + + I was not sure if the None values are related to the spelling mistakes or something else but after the ai's answer i noticed where is the problem was exactly in my code + --> diff --git a/database.py b/database.py index 8322d17..08a043a 100644 --- a/database.py +++ b/database.py @@ -25,16 +25,40 @@ def create_tables(conn: sqlite3.Connection) -> None: + UNIQUE(station, timestamp) constraint for upserts """ # TODO: use conn.execute() with CREATE TABLE IF NOT EXISTS statements - raise NotImplementedError + with conn: + conn.execute(""" + CREATE TABLE IF NOT EXISTS raw_weather( + id INTEGER PRIMARY KEY AUTOINCREMENT, + station TEXT NOT NULL, + timestamp TEXT NOT NULL, + temperature_c REAL NOT NULL, + humidity_pct INTEGER NOT NULL, + source TEXT NOT NULL, + ingested_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)""") + + conn.execute(""" + CREATE TABLE IF NOT EXISTS weather_readings( + id INTEGER PRIMARY KEY AUTOINCREMENT, + station TEXT NOT NULL, + timestamp TEXT NOT NULL, + temperature_c REAL NOT NULL, + humidity_pct INTEGER NOT NULL, + UNIQUE(station, timestamp))""") def insert_raw(conn: sqlite3.Connection, records: list[dict], source: str) -> None: """Insert raw records (before validation) into raw_weather. Use parameterized queries with placeholder syntax; do not build SQL via string formatting. - """ - # TODO: implement - raise NotImplementedError + """ + rows = [ + (r["station"], r["timestamp"], r["temperature_c"], r["humidity_pct"],source) + for r in records + ] + conn.executemany(""" + INSERT INTO raw_weather (station, timestamp, temperature_c, humidity_pct, source) + VALUES(?,?,?,?,?)""", + rows) def upsert_readings(conn: sqlite3.Connection, readings: list[WeatherReading]) -> None: @@ -43,11 +67,24 @@ def upsert_readings(conn: sqlite3.Connection, readings: list[WeatherReading]) -> Use the upsert pattern to handle duplicate (station, timestamp) pairs. Use parameterized queries. """ - # TODO: implement - raise NotImplementedError + rows = [ + (r.station, r.timestamp, r.temperature_c, r.humidity_pct) + for r in readings + ] + conn.executemany(""" + INSERT INTO weather_readings(station, timestamp, temperature_c , humidity_pct) + VALUES(?,?,?,?) + ON CONFLICT(station, timestamp)DO UPDATE SET + temperature_c=excluded.temperature_c, + humidity_pct=excluded.humidity_pct + """,rows) def count_readings(conn: sqlite3.Connection) -> int: """Return the total number of rows in weather_readings.""" - # TODO: implement - raise NotImplementedError + cursor=conn.execute("SELECT COUNT(*) FROM weather_readings") + row = cursor.fetchone() + if row: + return row[0] + else: + return 0 diff --git a/ingest_api.py b/ingest_api.py index 66eb32c..2e1723c 100644 --- a/ingest_api.py +++ b/ingest_api.py @@ -18,8 +18,33 @@ def fetch_with_retry(url: str, params: dict, max_retries: int = 3, timeout: int Fail immediately on: 4xx status codes. Log each retry attempt with the error and delay. """ - # TODO: implement retry loop with exponential backoff - raise NotImplementedError + for attempt in range(max_retries): + try: + response = requests.get( + url, + params=params, + timeout=timeout + ) + response.raise_for_status() + return response.json() + except(requests.exceptions.HTTPError) as e: + if response.status_code < 500: + logger.error(f"HTTP error {response.status_code} Cannot retry!") + raise e + if response.status_code >= 500: + logger.warning(f"Transient HTTP error {response.status_code} Retrying...") + if attempt == max_retries - 1: + raise e + wait_time = 2 ** attempt + logger.warning(f"Attempt {attempt + 1} failed with error: {response.status}") + time.sleep(wait_time) + except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e: + logger.warning(f"Transient network error: {e}. Retrying...") + if attempt == max_retries - 1: + raise e + wait_time = 2 ** attempt + logger.warning(f"Attempt {attempt + 1} failed with error: {e} Retrying in {wait_time} seconds...") + time.sleep(wait_time) def fetch_api_records() -> list[dict]: @@ -34,8 +59,21 @@ def fetch_api_records() -> list[dict]: "hourly": "temperature_2m,relative_humidity_2m", "forecast_days": 7, } - # TODO: - # - Call fetch_with_retry with API_URL and params - # - The API returns {"hourly": {"time": [...], "temperature_2m": [...], "relative_humidity_2m": [...]}} - # - Flatten to a list of dicts; set station="Open-Meteo Copenhagen" for all records - raise NotImplementedError + + try: + data = fetch_with_retry(API_URL,params) + hourly = data.get("hourly",{}) + records = [] + for i, timestamp in enumerate(hourly.get("time",[])): + records.append({ + "timestamp": timestamp, + "temperature_c": hourly.get("temperature_2m",[])[i], + "humidity_pct": hourly.get("relative_humidity_2m",[])[i], + "station": "Open-Meteo Copenhagen", + }) + + logger.info(f"Fetched {len(records)} weather records") + return records + except Exception: + return [] + diff --git a/ingest_files.py b/ingest_files.py index 8b010ff..d20ca98 100644 --- a/ingest_files.py +++ b/ingest_files.py @@ -16,5 +16,23 @@ def read_csv_records(path: Path) -> list[dict]: - Convert temperature_c to float and humidity_pct to int where possible. - Leave unconvertible values (e.g. "N/A", "") as-is so validation can catch them. """ - # TODO: implement CSV reading and normalization - raise NotImplementedError + with open(path, "r", newline="", encoding="utf-8") as file: + reader = csv.DictReader(file) + normalize_row =[] + for row in reader: + try: + temperature_c = float(row.get("temperature_c")) + except (ValueError, TypeError): + temperature_c = row.get("temperature_c") + try: + humidity_pct = int(row.get("humidity_pct")) + except (ValueError, TypeError): + humidity_pct = row.get("humidity_pct") + normalize_row.append({ + "station": row.get("station"), + "timestamp": row.get("timestamp"), + "temperature_c": temperature_c, + "humidity_pct": humidity_pct + }) + return normalize_row + diff --git a/models.py b/models.py index cae42fe..8c56628 100644 --- a/models.py +++ b/models.py @@ -13,5 +13,7 @@ class WeatherReading(BaseModel): @field_validator("station") @classmethod def clean_station(cls, v: str) -> str: - # TODO: strip whitespace and convert to title case - raise NotImplementedError + cleaned= v.strip().title() + if not cleaned: + raise ValueError("Station name shouldn't be empty!") + return cleaned diff --git a/output/azure_compare.md b/output/azure_compare.md index 22395b3..945bca3 100644 --- a/output/azure_compare.md +++ b/output/azure_compare.md @@ -4,12 +4,12 @@ Fill in each section below (2-3 sentences each) after completing the Task 7 step ## Auth - + ## Schema verbosity - + ## api-version in the URL - + diff --git a/pipeline.py b/pipeline.py index d8dc710..e62c61f 100644 --- a/pipeline.py +++ b/pipeline.py @@ -17,17 +17,31 @@ def run_pipeline() -> None: OUTPUT_DIR.mkdir(exist_ok=True) - - # TODO — implement each step in order: - # - # 1. Fetch records from Open-Meteo API using fetch_api_records() - # 2. Read records from CSV using read_csv_records(CSV_PATH) - # 3. Open a DB connection, create tables, insert all raw records (both sources) - # 4. Validate all records — collect valid WeatherReading objects and error dicts - # 5. Upsert valid records into weather_readings - # 6. Save error dicts as JSON to output/error_report.json - # 7. Print the pipeline summary in the format below. - # + api_fetching = fetch_api_records() + csv_reading = read_csv_records(CSV_PATH) + db_conn = get_connection() + create_tables(db_conn) + insert_raw(db_conn, api_fetching, source="API") + insert_raw(db_conn, csv_reading, source="CSV") + db_conn.commit() + api_valid, api_errors = validate_records(api_fetching, source="API") + csv_valid, csv_errors = validate_records(csv_reading, source="CSV") + all_valid_readings = api_valid + csv_valid + all_errors = api_errors + csv_errors + upsert_readings(db_conn, all_valid_readings) + db_conn.commit() + with open(OUTPUT_DIR / "error_report.json", "w") as f: + json.dump(all_errors, f, indent=2) + total_in_db = count_readings(db_conn) + print("=== Pipeline Summary ===") + print(f"API records fetched: {len(api_fetching)}") + print(f"CSV records read: {len(csv_reading)}") + print(f"Total raw records: {len(api_fetching) + len(csv_reading)}") + print(f"Valid records: {len(all_valid_readings)}") + print(f"Invalid records: {len(all_errors)}") + print(f"Records in database: {total_in_db}") + print(f"Error report: {OUTPUT_DIR / 'error_report.json'}") + db_conn.close() # Note: the API count varies by time of day (Open-Meteo returns up to 168 hourly # records for 7 forecast days; the exact number depends on the current UTC hour). # The CSV contributes 6 invalid records and 4 valid ones; the duplicate Copenhagen @@ -43,8 +57,6 @@ def run_pipeline() -> None: # Records in database: 169 # Error report: output/error_report.json - raise NotImplementedError - if __name__ == "__main__": logging.basicConfig(level=logging.INFO) diff --git a/test_azure.py b/test_azure.py new file mode 100644 index 0000000..21ab957 --- /dev/null +++ b/test_azure.py @@ -0,0 +1,29 @@ +import requests +import json +from pathlib import Path +import os +from dotenv import load_dotenv + +load_dotenv() +token = os.environ.get("AZURE_TOKEN") +subscription_id = os.environ.get("AZURE_SUBSCRIPTION_ID") + +OUTPUT_DIR = Path("output") +OUTPUT_DIR.mkdir(exist_ok=True) + +url = f"https://management.azure.com/subscriptions/{subscription_id}/resourcegroups?api-version=2024-03-01" +headers = {"Authorization": f"Bearer {token}"} + +response = requests.get(url, headers=headers, timeout=10) +response.raise_for_status() + +#for rg in response.json()["value"]: + #print(f"{rg['name']}: {rg['location']}") + + +azure_data = response.json() +output_file = OUTPUT_DIR / "azure_resource_groups.json" +with open(output_file, "w") as f: + json.dump(azure_data, f, indent=2) + +print(f"Successfully saved full response to {output_file}") diff --git a/validate.py b/validate.py index ada2716..456f643 100644 --- a/validate.py +++ b/validate.py @@ -19,4 +19,19 @@ def validate_records( error_details - the Pydantic error list (ValidationError.errors()) """ # TODO: iterate over records, try WeatherReading(**record), accumulate results - raise NotImplementedError + valid_list = [] + error_list = [] + for i, record in enumerate(records): + try: + valid_record= WeatherReading(**record) + except ValidationError as e: + error_dict = { + "index": i, + "source": source, + "raw_record": record, + "error_details": e.errors() + } + error_list.append(error_dict) + else: + valid_list.append(valid_record) + return valid_list, error_list From 7a3347b217107c756eda33e4f6d90796cb89dbb8 Mon Sep 17 00:00:00 2001 From: Mareh AboGhanem <85366580+mareh-aboghanem@users.noreply.github.com> Date: Sat, 30 May 2026 14:04:50 +0200 Subject: [PATCH 2/2] Refactor weather data extraction in ingest_api.py --- ingest_api.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/ingest_api.py b/ingest_api.py index 2e1723c..f8c7d38 100644 --- a/ingest_api.py +++ b/ingest_api.py @@ -64,11 +64,13 @@ def fetch_api_records() -> list[dict]: data = fetch_with_retry(API_URL,params) hourly = data.get("hourly",{}) records = [] + temperatures = hourly.get("temperature_2m", []) + humidities = hourly.get("relative_humidity_2m", []) for i, timestamp in enumerate(hourly.get("time",[])): records.append({ "timestamp": timestamp, - "temperature_c": hourly.get("temperature_2m",[])[i], - "humidity_pct": hourly.get("relative_humidity_2m",[])[i], + "temperature_c": temperatures[i], + "humidity_pct": humidities[i], "station": "Open-Meteo Copenhagen", })