From 0d0bd14c67db977896d73e18e967ea21edcc6b40 Mon Sep 17 00:00:00 2001 From: hannahwn Date: Wed, 20 May 2026 23:36:03 +0200 Subject: [PATCH] ingesting data example --- AI_DEBUG.md | 29 +++++++++++++++++++++++++++ database.py | 53 +++++++++++++++++++++++++++++++++++++++++++++---- ingest_api.py | 38 +++++++++++++++++++++++++++++++++-- ingest_files.py | 27 ++++++++++++++++++++++++- models.py | 24 ++++++++++++++++++++-- pipeline.py | 48 +++++++++++++++++++++++++++++++++++++++++++- validate.py | 15 +++++++++++++- 7 files changed, 223 insertions(+), 11 deletions(-) diff --git a/AI_DEBUG.md b/AI_DEBUG.md index edf34a5..e5f5f1a 100644 --- a/AI_DEBUG.md +++ b/AI_DEBUG.md @@ -9,18 +9,47 @@ Document the debugging session below. If everything worked first try, introduce +Traceback (most recent call last): + File "D:\a\_work\1\s\build_scripts\windows\artifacts\cli\Lib\site-packages\knack/cli.py", line 233, in invoke + File "D:\a\_work\1\s\build_scripts\windows\artifacts\cli\Lib\site-packages\azure/cli/core/commands/__init__.py", line 677, in execute + File "D:\a\_work\1\s\build_scripts\windows\artifacts\cli\Lib\site-packages\azure/cli/core/commands/__init__.py", line 820, in _run_jobs_serially + File "D:\a\_work\1\s\build_scripts\windows\artifacts\cli\Lib\site-packages\azure/cli/core/commands/__init__.py", line 789, in _run_job + File "D:\a\_work\1\s\build_scripts\windows\artifacts\cli\Lib\site-packages\azure/cli/core/commands/__init__.py", line 335, in __call__ + File "D:\a\_work\1\s\build_scripts\windows\artifacts\cli\Lib\site-packages\azure/cli/core/commands/command_operation.py", line 120, in handler + File "D:\a\_work\1\s\build_scripts\windows\artifacts\cli\Lib\site-packages\azure/cli/command_modules/profile/custom.py", line 215, in login + File "D:\a\_work\1\s\build_scripts\windows\artifacts\cli\Lib\site-packages\azure/cli/core/_profile.py", line 177, in login + File "D:\a\_work\1\s\build_scripts\windows\artifacts\cli\Lib\site-packages\azure/cli/core/auth/identity.py", line 175, in login_with_device_code + File "D:\a\_work\1\s\build_scripts\windows\artifacts\cli\Lib\site-packages\azure/cli/core/auth/identity.py", line 125, in _msal_app + File "D:\a\_work\1\s\build_scripts\windows\artifacts\cli\Lib\site-packages\msal/application.py", line 2090, in __init__ + File "D:\a\_work\1\s\build_scripts\windows\artifacts\cli\Lib\site-packages\msal/application.py", line 649, in __init__ + File "D:\a\_work\1\s\build_scripts\windows\artifacts\cli\Lib\site-packages\msal/authority.py", line 115, in __init__ +ValueError: Unable to get authority configuration for https://login.microsoftonline.com/07a14c4e-d88f-42f7-83b3-13af7e57ff3d. Authority would typically be in a format of https://login.microsoftonline.com/your_tenant or https://tenant_name.ciamlogin.com or https://tenant_name.b2clogin.com/tenant.onmicrosoft.com/policy. Also please double check your tenant name or GUID is correct. +To check existing issues, please visit: https://github.com/Azure/azure-cli/issues +Please run 'az login' to setup account. +Please run 'az login' to setup account. + + ## The Prompt + How to solve this? + ## The Solution + The error says the tenant ID is wrong. There is a typo in it! + ## Reflection + + + + + I have two accounts under my email and they seem to cause conflict when i try to login diff --git a/database.py b/database.py index 8322d17..05f78bf 100644 --- a/database.py +++ b/database.py @@ -25,7 +25,23 @@ 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 + conn.execute("""CREATE TABLE IF NOT EXISTS raw_weather ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + station TEXT, + timestamp TEXT, + temperature_c TEXT, + humidity_pct TEXT, + source TEXT, + ingested_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP +)""") + conn.execute("""CREATE TABLE IF NOT EXISTS weather_readings ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + station TEXT, + timestamp TEXT, + temperature_c REAL, + humidity_pct INTEGER, + UNIQUE(station, timestamp) +)""") def insert_raw(conn: sqlite3.Connection, records: list[dict], source: str) -> None: @@ -34,7 +50,19 @@ def insert_raw(conn: sqlite3.Connection, records: list[dict], source: str) -> No Use parameterized queries with placeholder syntax; do not build SQL via string formatting. """ # TODO: implement - raise NotImplementedError + for record in records: + conn.execute( + """INSERT INTO raw_weather (station, timestamp, temperature_c, humidity_pct, source) + VALUES (?, ?, ?, ?, ?)""", + ( + record.get("station"), + record.get("timestamp"), + record.get("temperature_c"), + record.get("humidity_pct"), + source, + ) + ) + def upsert_readings(conn: sqlite3.Connection, readings: list[WeatherReading]) -> None: @@ -44,10 +72,27 @@ def upsert_readings(conn: sqlite3.Connection, readings: list[WeatherReading]) -> Use parameterized queries. """ # TODO: implement - raise NotImplementedError + for reading in readings: + conn.execute( + """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""", + ( + reading.station, + reading.timestamp, + reading.temperature_c, + reading.humidity_pct, + ) + ) + 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") + count = cursor.fetchone()[0] + return count + diff --git a/ingest_api.py b/ingest_api.py index 66eb32c..fc03d6a 100644 --- a/ingest_api.py +++ b/ingest_api.py @@ -19,7 +19,22 @@ def fetch_with_retry(url: str, params: dict, max_retries: int = 3, timeout: int Log each retry attempt with the error and delay. """ # TODO: implement retry loop with exponential backoff - raise NotImplementedError + for attempt in range(1, max_retries + 1): + try: + response = requests.get(url, params=params, timeout=timeout) + if response.status_code >= 500: + raise requests.HTTPError(f"Server error: {response.status_code}") + elif response.status_code >= 400: + response.raise_for_status() # Will raise HTTPError for 4xx + return response.json() + except (requests.ConnectionError, requests.Timeout, requests.HTTPError) as e: + if attempt == max_retries or isinstance(e, requests.HTTPError) and e.response.status_code < 500: + logger.error(f"Request failed after {attempt} attempts: {e}") + raise + delay = 2 ** (attempt - 1) # Exponential backoff: 1s, 2s, 4s... + logger.warning(f"Request error: {e}. Retrying in {delay} seconds... (Attempt {attempt}/{max_retries})") + time.sleep(delay) + raise RuntimeError("Exceeded maximum retry attempts") def fetch_api_records() -> list[dict]: @@ -38,4 +53,23 @@ def fetch_api_records() -> list[dict]: # - 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 + for attempt in range(1, 4): + try: + data = fetch_with_retry(API_URL, params) + hourly = data.get("hourly", {}) + times = hourly.get("time", []) + temps = hourly.get("temperature_2m", []) + hums = hourly.get("relative_humidity_2m", []) + records = [] + for time_str, temp, hum in zip(times, temps, hums): + records.append({ + "station": "Open-Meteo Copenhagen", + "timestamp": time_str, + "temperature_c": temp, + "humidity_pct": hum, + }) + return records + except Exception as e: + logger.error(f"Failed to fetch API records: {e}") + return [] + raise RuntimeError("Exceeded maximum retry attempts") diff --git a/ingest_files.py b/ingest_files.py index 8b010ff..dd5f5f6 100644 --- a/ingest_files.py +++ b/ingest_files.py @@ -5,6 +5,20 @@ from pathlib import Path +def try_parse_float(value: str): + try: + return float(value) + except ValueError: + return value + + +def try_parse_int(value: str): + try: + return int(value) + except ValueError: + return value + + def read_csv_records(path: Path) -> list[dict]: """Read weather_stations.csv and return normalized records. @@ -17,4 +31,15 @@ def read_csv_records(path: Path) -> list[dict]: - Leave unconvertible values (e.g. "N/A", "") as-is so validation can catch them. """ # TODO: implement CSV reading and normalization - raise NotImplementedError + with path.open(newline="", encoding="utf-8") as f: + reader = csv.DictReader(f) + records = [] + for row in reader: + record = { + "station": row.get("station", "").strip(), + "timestamp": row.get("timestamp", "").strip(), + "temperature_c": try_parse_float(row.get("temperature_c", "").strip()), + "humidity_pct": try_parse_int(row.get("humidity_pct", "").strip()), + } + records.append(record) + return records \ No newline at end of file diff --git a/models.py b/models.py index cae42fe..256766d 100644 --- a/models.py +++ b/models.py @@ -1,7 +1,7 @@ # Step 1 — Task 4: Pydantic Validation # Define the WeatherReading model that every ingested record must pass. # Both the API and CSV data flow through this model before reaching the database. -from pydantic import BaseModel, Field, field_validator +from pydantic import BaseModel, Field, field_validator, ValidationError class WeatherReading(BaseModel): @@ -14,4 +14,24 @@ class WeatherReading(BaseModel): @classmethod def clean_station(cls, v: str) -> str: # TODO: strip whitespace and convert to title case - raise NotImplementedError + v = v.strip() + v = v.title() + return v + +def validate_readings(raw_records: list[dict]) -> tuple[list[WeatherReading], list[dict]]: + """Validate a list of raw records, returning valid records and errors.""" + valid = [] + errors = [] + + for i, record in enumerate(raw_records): + try: + reading = WeatherReading(**record) + valid.append(reading) + except ValidationError as e: + errors.append({ + "index": i, + "record": record, + "errors": e.errors(), + }) + + return valid, errors \ No newline at end of file diff --git a/pipeline.py b/pipeline.py index d8dc710..85371ec 100644 --- a/pipeline.py +++ b/pipeline.py @@ -18,6 +18,53 @@ def run_pipeline() -> None: OUTPUT_DIR.mkdir(exist_ok=True) + # step 1 - fetch from API + api_records = fetch_api_records() + + # step 2 - read from CSV + csv_records = read_csv_records(CSV_PATH) + + # step 3 - combine both + all_records = api_records + csv_records + + # step 4 - open database once + conn = get_connection() + create_tables(conn) + + # step 5 - insert all raw records + insert_raw(conn, api_records, "api") + insert_raw(conn, csv_records, "csv") + conn.commit() + + # step 6 - validate all records + valid_api, errors_api = validate_records(api_records, "api") + valid_csv, errors_csv = validate_records(csv_records, "csv") + valid = valid_api + valid_csv + errors = errors_api + errors_csv + + # step 7 - upsert valid records + upsert_readings(conn, valid) + conn.commit() + + # step 8 - save error report + error_report_path = OUTPUT_DIR / "error_report.json" + with error_report_path.open("w", encoding="utf-8") as f: + json.dump(errors, f, indent=2, default=str) + + # step 9 - print summary + print("=== Pipeline Summary ===") + print(f"API records fetched: {len(api_records)}") + print(f"CSV records read: {len(csv_records)}") + print(f"Total raw records: {len(all_records)}") + print(f"Valid records: {len(valid)}") + print(f"Invalid records: {len(errors)}") + print(f"Records in database: {count_readings(conn)}") + print(f"Error report: {error_report_path}") + + conn.close() + + + # TODO — implement each step in order: # # 1. Fetch records from Open-Meteo API using fetch_api_records() @@ -43,7 +90,6 @@ def run_pipeline() -> None: # Records in database: 169 # Error report: output/error_report.json - raise NotImplementedError if __name__ == "__main__": diff --git a/validate.py b/validate.py index ada2716..bbacec7 100644 --- a/validate.py +++ b/validate.py @@ -19,4 +19,17 @@ def validate_records( error_details - the Pydantic error list (ValidationError.errors()) """ # TODO: iterate over records, try WeatherReading(**record), accumulate results - raise NotImplementedError + valid_records = [] + error_records = [] + for i, record in enumerate(records): + try: + validated_record = WeatherReading(**record) + valid_records.append(validated_record) + except ValidationError as e: + error_records.append({ + "index": i, + "source": source, + "raw_record": record, + "error_details": e.errors() + }) + return valid_records, error_records