From 2af4986b76a5c0a15867f82b699b8ae24ac49f5d Mon Sep 17 00:00:00 2001 From: ZeroQuest96 Date: Fri, 19 Jun 2026 09:08:20 -0500 Subject: [PATCH 01/13] chore: update sources.yml to include web csv --- config/sources.yml | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/config/sources.yml b/config/sources.yml index 9e5d1f2..3dbeee7 100644 --- a/config/sources.yml +++ b/config/sources.yml @@ -4,6 +4,28 @@ defaults: on_conflict: upsert sources: + - name: noaa_test + type: http_csv + url: https://www.ncei.noaa.gov/data/global-summary-of-the-month/access/USC00129222.csv + target_table: stg_noaa_test + + pk: [station, date] + + schema: + station: str + name: str + date: datetime + prcp: float + snow: float + tmax: float + tmin: float + awnd: float + + rules: + - prcp >= 0 + - snow >= 0 + - tmax >= tmin + - name: noaa_monthly_sample type: csv path: sample_data/small_sample.csv #sample_data/gsom_sample_csv.csv From 810f59beec4db4d5e81e83a7db0c59edbf97dbe8 Mon Sep 17 00:00:00 2001 From: ZeroQuest96 Date: Fri, 19 Jun 2026 09:08:51 -0500 Subject: [PATCH 02/13] feat: add source ingestion for web csv --- src/readers/readers.py | 41 ++++++++++++++++++++++++++++++++++++++++- 1 file changed, 40 insertions(+), 1 deletion(-) diff --git a/src/readers/readers.py b/src/readers/readers.py index ad33063..9582813 100644 --- a/src/readers/readers.py +++ b/src/readers/readers.py @@ -2,8 +2,11 @@ import logging import os import pandas as pd +import requests -logger = logging.getLogger(__name__) +from io import StringIO + +from loggers.logging_config import logger def read_input(source): @@ -16,6 +19,8 @@ def read_input(source): return read_csv(source) elif source["type"] == "json": return read_json(source) + elif source["type"] == "http_csv": + return read_http_csv(source) else: raise ValueError(f"Unsupported source type {source['type']}") @@ -92,3 +97,37 @@ def read_json(source): logger.info(f"JSON file read from: {path}") return df + +def read_http_csv(source): + """ + Reads one or more remote CSV files over HTTP(S) baseed on the config. + """ + + urls = [] + + if "url" in source: + urls = [source["url"]] + elif "urls" in source: + urls = source["urls"] + else: + raise ValueError("HTTP CSV source must define either 'url' or 'urls'.") + + frames = [] + + for url in urls: + logger.info(f"Downloading CSV from {url}") + + try: + response = requests.get(url, timeout=30) + response.raise_for_status() + + df = pd.read_csv(StringIO(response.text)) + frames.append(df) + + except requests.RequestException as e: + raise ConnectionError(f"Failed to download CSV from: {url}") from e + + except pd.errors.ParserError as e: + raise ValueError(f"FAiled to parse CSV from: {url}") from e + + return pd.concat(frames, ignore_index=True) From dc469b00df19c8b20ebd3d1f24874b5d0e2cbecd Mon Sep 17 00:00:00 2001 From: ZeroQuest96 Date: Fri, 19 Jun 2026 09:09:54 -0500 Subject: [PATCH 03/13] fix: allow non-primary key entries to be null --- src/validators/validators.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/src/validators/validators.py b/src/validators/validators.py index 88153ee..aa33011 100644 --- a/src/validators/validators.py +++ b/src/validators/validators.py @@ -26,7 +26,8 @@ def apply_schema_casts(df, schema, source_name): # Config validation if col not in df.columns: # Possibly change to a warning log later - raise KeyError(f"Schema column missing in dataframe: {col}") + logger.warning(f"[{source_name}] Missing column '{col}' filling with NA") + df[col] = pd.Series([pd.NA] * len(df)) try: target_type = PANDAS_TYPE_MAP[col_type] @@ -41,9 +42,14 @@ def apply_schema_casts(df, schema, source_name): converted = pd.to_datetime(df[col], errors="coerce") except Exception as e: raise ValueError(f"Failed to convert column '{col}' to datetime") from e + elif col_type == "str": + try: + converted = df[col].astype(col_type) + except Exception as e: + raise ValueError(f"Failed to cast column {col} to {target_type}") else: try: - converted = df[col].astype(target_type) + converted = pd.to_numeric(df[col], errors="coerce") except Exception as e: raise ValueError(f"Failed to cast column {col} to {target_type}") from e From 1ae7a272c62020ccb090974bca087a2d7cbcbcb7 Mon Sep 17 00:00:00 2001 From: ZeroQuest96 Date: Fri, 19 Jun 2026 10:01:23 -0500 Subject: [PATCH 04/13] feat: add source resolver for giving specific sources specific rules --- src/sources/resolver.py | 49 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 49 insertions(+) create mode 100644 src/sources/resolver.py diff --git a/src/sources/resolver.py b/src/sources/resolver.py new file mode 100644 index 0000000..dd2926c --- /dev/null +++ b/src/sources/resolver.py @@ -0,0 +1,49 @@ +from loggers.logging_config import logger + +def resolve_source(source): + """ + Turn high-level dataset definitions into executable ingestion configs. + """ + + source_type = source.get("type") + + if source_type == "noaa_gsom": + return resolve_noaa_gsom(source) + + # We passthrough for generic sources + return source + +def resolve_noaa_gsom(source): + """ + Expands NOAA GSOM config into http_csv ingestion config + """ + base_url = source.get("base_url") + stations = source.get("stations", []) + + if not base_url: + raise ValueError("noaa_gsom requires 'base_url'") + if not stations: + raise ValueError("noaa_gsom requires 'stations'") + + logger.info(f"Resolving NOAA GSOM source for {len(stations)} stations") + + urls = [ + f"{base_url.rstrip('/')}/{station}.csv" + for station in stations + ] + + # Return normalized config for existing pipeline + resolved = { + "type": "http_csv", + "urls": urls, + + "name": source.get("name"), + "target_table": source.get("target_table"), + "pk": source.get("pk"), + "schema": source.get("schema"), + "rules": source.get("rules"), + + "time_filter": source.get("time_filter") + } + + return resolved \ No newline at end of file From 4c799916fce83063e420c3aea3e2a5ddbda5664d Mon Sep 17 00:00:00 2001 From: ZeroQuest96 Date: Fri, 19 Jun 2026 10:02:12 -0500 Subject: [PATCH 05/13] feat: add ability to apply a time filter --- src/cleaners/cleaners.py | 28 ++++++++++++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/src/cleaners/cleaners.py b/src/cleaners/cleaners.py index 201f02d..eb789dd 100644 --- a/src/cleaners/cleaners.py +++ b/src/cleaners/cleaners.py @@ -30,3 +30,31 @@ def normalize_for_database(df): df = df.where(pd.notnull(df), None) return df + +def apply_time_filter(df, source): + """ + Filter for a specific time range. + """ + + time_filter = source.get("time_filter") + if not time_filter: + return df + + column = time_filter.get("column") + start = time_filter.get("start") + end = time_filter.get("end") + + if column not in df.columns: + raise ValueError(f"Time filter column '{column}' not in dataframe") + + logger.info(f"Applying time filter on {column}: {start} -> {end}") + + df[column] = pd.to_datetime(df[column], errors="coerce") + + if start: + df = df[df[column] >= pd.to_datetime(start)] + + if end: + df = df[df[column] <= pd.to_datetime(end)] + + return df \ No newline at end of file From 56b77d4784f88e34bfcc2572a42614dac73f7afe Mon Sep 17 00:00:00 2001 From: ZeroQuest96 Date: Fri, 19 Jun 2026 10:04:06 -0500 Subject: [PATCH 06/13] fix: update web csv reader to work with new source flow --- src/readers/readers.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/readers/readers.py b/src/readers/readers.py index 9582813..78e2a88 100644 --- a/src/readers/readers.py +++ b/src/readers/readers.py @@ -1,5 +1,4 @@ import json -import logging import os import pandas as pd import requests @@ -8,7 +7,6 @@ from loggers.logging_config import logger - def read_input(source): """ Reads the source input based on the configuration From 7ba159913be0ee2ad1eb0449fa77628a35d43dbb Mon Sep 17 00:00:00 2001 From: ZeroQuest96 Date: Fri, 19 Jun 2026 10:05:07 -0500 Subject: [PATCH 07/13] chore: update sources.yml to include time filtering --- config/sources.yml | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/config/sources.yml b/config/sources.yml index 3dbeee7..f934015 100644 --- a/config/sources.yml +++ b/config/sources.yml @@ -5,8 +5,14 @@ defaults: sources: - name: noaa_test - type: http_csv - url: https://www.ncei.noaa.gov/data/global-summary-of-the-month/access/USC00129222.csv + type: noaa_gsom + base_url: https://www.ncei.noaa.gov/data/global-summary-of-the-month/access + + stations: + - USC00218450 + - USC00129222 + - USW00003812 + target_table: stg_noaa_test pk: [station, date] @@ -26,6 +32,11 @@ sources: - snow >= 0 - tmax >= tmin + time_filter: + column: date + start: 2016-01-01 + end: 2020-12-31 + - name: noaa_monthly_sample type: csv path: sample_data/small_sample.csv #sample_data/gsom_sample_csv.csv From 339b0d40877f08fb7d2a6a6b46fcfb6dc9367cfe Mon Sep 17 00:00:00 2001 From: ZeroQuest96 Date: Fri, 19 Jun 2026 10:05:41 -0500 Subject: [PATCH 08/13] feat: add source resolution and time filtering to pipeline --- src/main.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/main.py b/src/main.py index aa9e37b..e95c1b4 100644 --- a/src/main.py +++ b/src/main.py @@ -3,8 +3,10 @@ from config.config_loader import load_config from database.database import database_setup, create_schema +from sources.resolver import resolve_source + from readers.readers import read_input -from cleaners.cleaners import normalize_columns, normalize_for_database +from cleaners.cleaners import normalize_columns, normalize_for_database, apply_time_filter from validators.validators import ( apply_schema_casts, @@ -27,9 +29,11 @@ def run_source(connection, source, defaults): Runs the ETL pipeline for a given source. """ log_source_start(str(source["name"])) + source = resolve_source(source) df = read_input(source) df = normalize_columns(df) df, rejects = apply_schema_casts(df, source["schema"], source["name"]) + df = apply_time_filter(df, source) df = df[list(source["schema"].keys())] df, enforce_required_rejects = enforce_required(df, source["pk"], source["name"]) rejects += enforce_required_rejects From 85dc0970fe8a01f56712cb1bab78360c80a782d0 Mon Sep 17 00:00:00 2001 From: ZeroQuest96 Date: Mon, 22 Jun 2026 08:45:05 -0500 Subject: [PATCH 09/13] feat: add open meteo source resolution --- src/sources/__init__.py | 0 src/sources/resolver.py | 67 ++++++++++++++++++++++++++++++++++++----- 2 files changed, 59 insertions(+), 8 deletions(-) create mode 100644 src/sources/__init__.py diff --git a/src/sources/__init__.py b/src/sources/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/sources/resolver.py b/src/sources/resolver.py index dd2926c..162c309 100644 --- a/src/sources/resolver.py +++ b/src/sources/resolver.py @@ -9,7 +9,8 @@ def resolve_source(source): if source_type == "noaa_gsom": return resolve_noaa_gsom(source) - + elif source_type == "open_meteo": + return resolve_open_meteo(source) # We passthrough for generic sources return source @@ -20,17 +21,24 @@ def resolve_noaa_gsom(source): base_url = source.get("base_url") stations = source.get("stations", []) - if not base_url: + if not isinstance(base_url, str) or not base_url.strip(): raise ValueError("noaa_gsom requires 'base_url'") - if not stations: + if not stations or not isinstance(stations, list): raise ValueError("noaa_gsom requires 'stations'") logger.info(f"Resolving NOAA GSOM source for {len(stations)} stations") - urls = [ - f"{base_url.rstrip('/')}/{station}.csv" - for station in stations - ] + urls = [] + + for station in stations: + if not isinstance(station, str) or not station.strip(): + logger.warning(f"Skipping invalid station id: {station}") + continue + + urls.append(f"{base_url.rstrip('/')}/{station}.csv") + + if not urls: + raise ValueError("noaa_gsom produced no valid URLs") # Return normalized config for existing pipeline resolved = { @@ -46,4 +54,47 @@ def resolve_noaa_gsom(source): "time_filter": source.get("time_filter") } - return resolved \ No newline at end of file + return resolved + +def resolve_open_meteo(source): + """ + Expands Open Meteo config into http_json config + """ + base_url = source["base_url"] + + if not base_url: + raise ValueError("open_meteo requires 'base_url'") + + params = { + "latitude": source["latitude"], + "longitude": source["longitude"], + "start_date": source["start_date"], + "end_date": source["end_date"], + "timezone": "UTC", + } + + hourly = source["params"]["hourly"] + params["hourly"] = ",".join(hourly) + + url = base_url + "?" + "&".join( + f"{key}={value}" for key, value in params.items() + ) + + logger.info( + f"Resolving source: {source.get('name')} -> http_json | " + f"lat={source.get('latitude')} lon={source.get('longitude')} " + f"range={source.get('start_date')}:{source.get('end_date')}" + ) + + return { + "type": "http_json", + "url": url, + + "name": source["name"], + "target_table": source["target_table"], + "pk": source["pk"], + "schema": source["schema"], + "rules": source["rules"], + + "json_root": "hourly", + } \ No newline at end of file From d7721cd8e0037d3e9faea5c60e6ddd1eb6a55045 Mon Sep 17 00:00:00 2001 From: ZeroQuest96 Date: Mon, 22 Jun 2026 08:45:45 -0500 Subject: [PATCH 10/13] feat: add http_json reading --- src/readers/readers.py | 65 +++++++++++++++++++++++++++++++++++++++++- 1 file changed, 64 insertions(+), 1 deletion(-) diff --git a/src/readers/readers.py b/src/readers/readers.py index 78e2a88..b0fa3e5 100644 --- a/src/readers/readers.py +++ b/src/readers/readers.py @@ -19,6 +19,8 @@ def read_input(source): return read_json(source) elif source["type"] == "http_csv": return read_http_csv(source) + elif source["type"] == "http_json": + return read_http_json(source) else: raise ValueError(f"Unsupported source type {source['type']}") @@ -107,9 +109,13 @@ def read_http_csv(source): urls = [source["url"]] elif "urls" in source: urls = source["urls"] - else: + + if not urls: raise ValueError("HTTP CSV source must define either 'url' or 'urls'.") + if isinstance(urls, str): + urls = [urls] + frames = [] for url in urls: @@ -123,9 +129,66 @@ def read_http_csv(source): frames.append(df) except requests.RequestException as e: + logger.error(f"HTTP failure for CSV source: {url}") raise ConnectionError(f"Failed to download CSV from: {url}") from e except pd.errors.ParserError as e: + logger.error(f"CSV parse failure: {url}") raise ValueError(f"FAiled to parse CSV from: {url}") from e + if not frames: + raise ValueError("No CSV frames were successfully loaded from source.") + + logger.info(f"Loaded HTTP CSV: {len(frames)} files combined") + return pd.concat(frames, ignore_index=True) + +def read_http_json(source): + """ + Reads json over http(s) + """ + url = source["url"] + + if not url: + raise ValueError("HTTP JSON source missing 'url'.") + + logger.info(f"Fetching JSON from {url}") + + try: + response = requests.get(url, timeout=30) + response.raise_for_status() + + data = response.json() + + except requests.RequestException as e: + logger.error(f"HTTP failure for JSON source: {url}") + raise ConnectionError(f"Failed to fetch JSON from: {url}") from e + + except ValueError as e: + logger.error(f"Invalid JSON reponse: {url}") + raise ValueError(f"Failed to parse JSON from: {url}") from e + + json_root = source.get("json_root") + + try: + # Case for Open-Meteo + if json_root: + root = data[json_root] + df = pd.DataFrame(root) + else: + df = pd.DataFrame([data]) + except Exception as e: + logger.error(f"Failed to normalize JSON structure: {url}") + raise ValueError(f"Invalid JSON structure for: {url}") from e + + if isinstance(data, dict): + if "latitude" in data: + df["latitude"] = data["latitude"] + + if "longitude" in data: + df["longitude"] = data["longitude"] + + logger.info(f"Loaded HTTP JSON rows: {len(df)}") + + return df + From 0bbd4791d785db16660e341639fc8bb508bf5116 Mon Sep 17 00:00:00 2001 From: ZeroQuest96 Date: Mon, 22 Jun 2026 08:46:21 -0500 Subject: [PATCH 11/13] chore: add source config for open meteo --- config/sources.yml | 59 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 59 insertions(+) diff --git a/config/sources.yml b/config/sources.yml index f934015..c419953 100644 --- a/config/sources.yml +++ b/config/sources.yml @@ -37,6 +37,65 @@ sources: start: 2016-01-01 end: 2020-12-31 + - name: open_meteo_api + + type: open_meteo + + base_url: https://archive-api.open-meteo.com/v1/archive + + latitude: 35.4319 + longitude: -82.5375 + + start_date: 2024-01-01 + end_date: 2025-12-31 + + params: + hourly: + - temperature_2m + - apparent_temperature + - precipitation + - snowfall + - wind_speed_10m + - wind_gusts_10m + - relative_humidity_2m + - cloud_cover + - pressure_msl + + target_table: stg_open_meteo + + pk: + - latitude + - longitude + - time + + schema: + latitude: float + longitude: float + time: datetime + + temperature_2m: float + apparent_temperature: float + precipitation: float + snowfall: float + + wind_speed_10m: float + wind_gusts_10m: float + + relative_humidity_2m: float + cloud_cover: float + pressure_msl: float + + rules: + - latitude >= -90 + - latitude <= 90 + - longitude >= -180 + - longitude <= 180 + - precipitation >= 0 + - snowfall >= 0 + - wind_speed_10m >= 0 + - wind_gusts_10m >= 0 + + - name: noaa_monthly_sample type: csv path: sample_data/small_sample.csv #sample_data/gsom_sample_csv.csv From 643dce26f2df71a4514f5be8dde3f48a3fc49ae1 Mon Sep 17 00:00:00 2001 From: ZeroQuest96 Date: Mon, 22 Jun 2026 08:47:10 -0500 Subject: [PATCH 12/13] feat: add tests for open meteo resolution and http json reading --- src/tests/test_reader.py | 155 +++++++++++++++++++++++++++++++++++- src/tests/test_resolver.py | 58 ++++++++++++++ src/tests/test_validator.py | 61 +++++++++++++- 3 files changed, 271 insertions(+), 3 deletions(-) create mode 100644 src/tests/test_resolver.py diff --git a/src/tests/test_reader.py b/src/tests/test_reader.py index 4673dfc..4e302b7 100644 --- a/src/tests/test_reader.py +++ b/src/tests/test_reader.py @@ -1,9 +1,17 @@ import json import pandas as pd import pytest +import requests -from readers.readers import read_input, read_csv, read_json +from unittest.mock import Mock +from readers.readers import ( + read_input, + read_csv, + read_json, + read_http_csv, + read_http_json +) def test_read_input_routes_to_csv(tmp_path, monkeypatch): file = tmp_path / "data.csv" @@ -115,3 +123,148 @@ def test_read_json_invalid_json(tmp_path, monkeypatch): with pytest.raises(ValueError): read_json(source) + +def test_read_http_csv(monkeypatch): + response = Mock() + response.text = "id,name\n1,bob\n2,alice" + response.raise_for_status.return_value = None + + monkeypatch.setattr( + "readers.readers.requests.get", + lambda url, timeout: response, + ) + + df = read_http_csv({ + "url": "https://example.com/test.csv" + }) + + assert len(df) == 2 + assert list(df.columns) == ["id", "name"] + +def test_read_http_csv_multiple_urls(monkeypatch): + + response = Mock() + response.text = "id\n1" + response.raise_for_status.return_value = None + + monkeypatch.setattr( + "readers.readers.requests.get", + lambda url, timeout: response, + ) + + df = read_http_csv({ + "urls": [ + "https://example.com/a.csv", + "https://example.com/b.csv" + ] + }) + + assert len(df) == 2 + + def test_read_http_csv_connection_error(monkeypatch): + + def fake_get(url, timeout): + raise requests.RequestException("network failure") + + monkeypatch.setattr( + "readers.readers.requests.get", + fake_get, + ) + + with pytest.raises(ConnectionError): + read_http_csv({ + "url": "https://example.com/test.csv" + }) + +def test_read_http_csv_missing_url(): + + with pytest.raises(ValueError): + read_http_csv({}) + +def test_read_http_json(monkeypatch): + + response = Mock() + + response.raise_for_status.return_value = None + response.json.return_value = { + "latitude": 35.43, + "longitude": -82.53, + "hourly": { + "time": [ + "2024-01-01T00:00", + "2024-01-01T01:00", + ], + "temperature_2m": [ + 12.5, + 13.1, + ], + }, + } + + monkeypatch.setattr( + "readers.readers.requests.get", + lambda url, timeout: response, + ) + + df = read_http_json({ + "url": "https://api.open-meteo.com", + "json_root": "hourly", + }) + + assert len(df) == 2 + assert "latitude" in df.columns + assert "longitude" in df.columns + assert "temperature_2m" in df.columns + +def test_read_http_json_connection_error(monkeypatch): + + def fake_get(url, timeout): + raise requests.RequestException("connection failure") + + monkeypatch.setattr( + "readers.readers.requests.get", + fake_get, + ) + + with pytest.raises(ConnectionError): + read_http_json({ + "url": "https://api.open-meteo.com" + }) + +def test_read_http_json_invalid_json(monkeypatch): + + response = Mock() + + response.raise_for_status.return_value = None + response.json.side_effect = ValueError("invalid json") + + monkeypatch.setattr( + "readers.readers.requests.get", + lambda url, timeout: response, + ) + + with pytest.raises(ValueError): + read_http_json({ + "url": "https://api.open-meto.com" + }) + +def test_read_http_json_invalid_root(monkeypatch): + + response = Mock() + + response.raise_for_status.return_value = None + response.json.return_value = { + "latitude": 35.43, + "longitude": -82.53, + } + + monkeypatch.setattr( + "readers.readers.requests.get", + lambda url, timeout: response, + ) + + with pytest.raises(ValueError): + read_http_json({ + "url": "https://api.open-meteo.com", + "json_root": "hourly", + }) \ No newline at end of file diff --git a/src/tests/test_resolver.py b/src/tests/test_resolver.py new file mode 100644 index 0000000..478af1e --- /dev/null +++ b/src/tests/test_resolver.py @@ -0,0 +1,58 @@ +import pytest +from sources.resolver import ( + resolve_noaa_gsom, + resolve_open_meteo, + resolve_source +) + +def test_noaa_resolver_expands_urls(): + source = { + "name": "noaa_test", + "type": "noaa_gsom", + "base_url": "https://example.com/data", + "stations": ["US1", "US2"], + "target_table": "stg", + "pk": ["station"], + "schema": {"station": "str"}, + "rules": [] + } + + resolved = resolve_noaa_gsom(source) + + assert resolved["type"] == "http_csv" + assert len(resolved["urls"]) == 2 + assert resolved["urls"][0].endswith("US1.csv") + +def test_noaa_missing_stations_raises(): + source = { + "name": "bad", + "type": "noaa_gsom", + "base_url": "x", + "stations": [] + } + + with pytest.raises(ValueError): + resolve_noaa_gsom(source) + +def test_open_meteo_url_build(): + source = { + "name": "open_meteo", + "type": "open_meteo", + "base_url": "https://archive-api.open-meteo.com/v1/archive", + "latitude": 10, + "longitude": 20, + "start_date": "2024-01-01", + "end_date": "2024-01-02", + "params": { + "hourly": ["temperature_2m"] + }, + "target_table": "t", + "pk": ["time"], + "schema": {}, + "rules": [] + } + + resolved = resolve_open_meteo(source) + + assert "temperature_2m" in resolved["url"] + assert "latitude=10" in resolved["url"] \ No newline at end of file diff --git a/src/tests/test_validator.py b/src/tests/test_validator.py index d0d6912..657a889 100644 --- a/src/tests/test_validator.py +++ b/src/tests/test_validator.py @@ -40,8 +40,11 @@ def test_apply_schema_casts_missing_column(): schema = {"date": "datetime"} - with pytest.raises(KeyError): - apply_schema_casts(df, schema, "test_source") + result, rejects = apply_schema_casts(df, schema, "test_source") + + assert "date" in result.columns + assert result["date"].isna().all() + assert len(rejects) == 0 def test_apply_schema_casts_invalid_type(): @@ -52,6 +55,60 @@ def test_apply_schema_casts_invalid_type(): with pytest.raises(ValueError): apply_schema_casts(df, schema, "test_source") +def test_apply_schema_casts_missing_float_column(): + df = pd.DataFrame({"station": ["ABC"]}) + + schema = {"awnd": "float"} + + result, rejects = apply_schema_casts(df, schema, "test_source") + + assert "awnd" in result.columns + assert result["awnd"].isna().all() + assert len(rejects) == 0 + +def test_apply_schema_casts_missing_string_column(): + df = pd.DataFrame({"station": ["ABC"]}) + + schema = {"name": "str"} + + result, rejects = apply_schema_casts(df, schema, "test_source") + + assert "name" in result.columns + assert result["name"].isna().all() + assert len(rejects) == 0 + +def test_apply_schema_casts_missing_int_column(): + df = pd.DataFrame({"station": ["ABC"]}) + + schema = {"elevation": "int"} + + result, rejects = apply_schema_casts(df, schema, "test_source") + + assert "elevation" in result.columns + assert result["elevation"].isna().all() + assert len(rejects) == 0 + +def test_apply_schema_casts_missing_datetime_column(): + df = pd.DataFrame({"station": ["ABC"]}) + + schema = {"date": "datetime"} + + result, rejects = apply_schema_casts(df, schema, "test_source") + + assert "date" in result.columns + assert result["date"].isna().all() + assert len(rejects) == 0 + +def test_apply_schema_casts_invalid_float_creates_reject(): + df = pd.DataFrame({"prcp": ["1.5", "abc"]}) + + schema = {"prcp": "float"} + + result, rejects = apply_schema_casts(df, schema, "test_source") + + assert pd.isna(result.loc[1, "prcp"]) + assert len(rejects) == 1 + assert rejects[0]["reason"] == "schema_cast_failed:prcp" def test_enforce_required_valid(): df = pd.DataFrame({"station": ["ABC"]}) From 608a5b821fc1205106686ef5849e2d1e92354163 Mon Sep 17 00:00:00 2001 From: ZeroQuest96 Date: Mon, 22 Jun 2026 08:50:37 -0500 Subject: [PATCH 13/13] chore: lint codebase for push --- config/sources.yml | 2 +- src/cleaners/cleaners.py | 5 +-- src/main.py | 6 +++- src/readers/readers.py | 20 ++++++----- src/sources/resolver.py | 19 +++++------ src/tests/test_reader.py | 65 ++++++++++++++++++------------------ src/tests/test_resolver.py | 26 +++++---------- src/tests/test_validator.py | 6 ++++ src/validators/validators.py | 2 +- 9 files changed, 77 insertions(+), 74 deletions(-) diff --git a/config/sources.yml b/config/sources.yml index c419953..09c6040 100644 --- a/config/sources.yml +++ b/config/sources.yml @@ -98,7 +98,7 @@ sources: - name: noaa_monthly_sample type: csv - path: sample_data/small_sample.csv #sample_data/gsom_sample_csv.csv + path: sample_data/small_sample.csv target_table: stg_noaa_monthly pk: [station, date] diff --git a/src/cleaners/cleaners.py b/src/cleaners/cleaners.py index eb789dd..d7ac3c7 100644 --- a/src/cleaners/cleaners.py +++ b/src/cleaners/cleaners.py @@ -31,6 +31,7 @@ def normalize_for_database(df): return df + def apply_time_filter(df, source): """ Filter for a specific time range. @@ -39,7 +40,7 @@ def apply_time_filter(df, source): time_filter = source.get("time_filter") if not time_filter: return df - + column = time_filter.get("column") start = time_filter.get("start") end = time_filter.get("end") @@ -57,4 +58,4 @@ def apply_time_filter(df, source): if end: df = df[df[column] <= pd.to_datetime(end)] - return df \ No newline at end of file + return df diff --git a/src/main.py b/src/main.py index e95c1b4..dc6bff5 100644 --- a/src/main.py +++ b/src/main.py @@ -6,7 +6,11 @@ from sources.resolver import resolve_source from readers.readers import read_input -from cleaners.cleaners import normalize_columns, normalize_for_database, apply_time_filter +from cleaners.cleaners import ( + normalize_columns, + normalize_for_database, + apply_time_filter, +) from validators.validators import ( apply_schema_casts, diff --git a/src/readers/readers.py b/src/readers/readers.py index b0fa3e5..356a869 100644 --- a/src/readers/readers.py +++ b/src/readers/readers.py @@ -7,6 +7,7 @@ from loggers.logging_config import logger + def read_input(source): """ Reads the source input based on the configuration @@ -98,6 +99,7 @@ def read_json(source): logger.info(f"JSON file read from: {path}") return df + def read_http_csv(source): """ Reads one or more remote CSV files over HTTP(S) baseed on the config. @@ -109,10 +111,10 @@ def read_http_csv(source): urls = [source["url"]] elif "urls" in source: urls = source["urls"] - + if not urls: raise ValueError("HTTP CSV source must define either 'url' or 'urls'.") - + if isinstance(urls, str): urls = [urls] @@ -131,18 +133,19 @@ def read_http_csv(source): except requests.RequestException as e: logger.error(f"HTTP failure for CSV source: {url}") raise ConnectionError(f"Failed to download CSV from: {url}") from e - + except pd.errors.ParserError as e: logger.error(f"CSV parse failure: {url}") raise ValueError(f"FAiled to parse CSV from: {url}") from e - + if not frames: raise ValueError("No CSV frames were successfully loaded from source.") - + logger.info(f"Loaded HTTP CSV: {len(frames)} files combined") return pd.concat(frames, ignore_index=True) + def read_http_json(source): """ Reads json over http(s) @@ -151,7 +154,7 @@ def read_http_json(source): if not url: raise ValueError("HTTP JSON source missing 'url'.") - + logger.info(f"Fetching JSON from {url}") try: @@ -159,11 +162,11 @@ def read_http_json(source): response.raise_for_status() data = response.json() - + except requests.RequestException as e: logger.error(f"HTTP failure for JSON source: {url}") raise ConnectionError(f"Failed to fetch JSON from: {url}") from e - + except ValueError as e: logger.error(f"Invalid JSON reponse: {url}") raise ValueError(f"Failed to parse JSON from: {url}") from e @@ -191,4 +194,3 @@ def read_http_json(source): logger.info(f"Loaded HTTP JSON rows: {len(df)}") return df - diff --git a/src/sources/resolver.py b/src/sources/resolver.py index 162c309..1ece30f 100644 --- a/src/sources/resolver.py +++ b/src/sources/resolver.py @@ -1,5 +1,6 @@ from loggers.logging_config import logger + def resolve_source(source): """ Turn high-level dataset definitions into executable ingestion configs. @@ -14,6 +15,7 @@ def resolve_source(source): # We passthrough for generic sources return source + def resolve_noaa_gsom(source): """ Expands NOAA GSOM config into http_csv ingestion config @@ -25,7 +27,7 @@ def resolve_noaa_gsom(source): raise ValueError("noaa_gsom requires 'base_url'") if not stations or not isinstance(stations, list): raise ValueError("noaa_gsom requires 'stations'") - + logger.info(f"Resolving NOAA GSOM source for {len(stations)} stations") urls = [] @@ -36,7 +38,7 @@ def resolve_noaa_gsom(source): continue urls.append(f"{base_url.rstrip('/')}/{station}.csv") - + if not urls: raise ValueError("noaa_gsom produced no valid URLs") @@ -44,18 +46,17 @@ def resolve_noaa_gsom(source): resolved = { "type": "http_csv", "urls": urls, - "name": source.get("name"), "target_table": source.get("target_table"), "pk": source.get("pk"), "schema": source.get("schema"), "rules": source.get("rules"), - - "time_filter": source.get("time_filter") + "time_filter": source.get("time_filter"), } return resolved + def resolve_open_meteo(source): """ Expands Open Meteo config into http_json config @@ -76,9 +77,7 @@ def resolve_open_meteo(source): hourly = source["params"]["hourly"] params["hourly"] = ",".join(hourly) - url = base_url + "?" + "&".join( - f"{key}={value}" for key, value in params.items() - ) + url = base_url + "?" + "&".join(f"{key}={value}" for key, value in params.items()) logger.info( f"Resolving source: {source.get('name')} -> http_json | " @@ -89,12 +88,10 @@ def resolve_open_meteo(source): return { "type": "http_json", "url": url, - "name": source["name"], "target_table": source["target_table"], "pk": source["pk"], "schema": source["schema"], "rules": source["rules"], - "json_root": "hourly", - } \ No newline at end of file + } diff --git a/src/tests/test_reader.py b/src/tests/test_reader.py index 4e302b7..8e73706 100644 --- a/src/tests/test_reader.py +++ b/src/tests/test_reader.py @@ -6,13 +6,14 @@ from unittest.mock import Mock from readers.readers import ( - read_input, - read_csv, + read_input, + read_csv, read_json, read_http_csv, - read_http_json + read_http_json, ) + def test_read_input_routes_to_csv(tmp_path, monkeypatch): file = tmp_path / "data.csv" file.write_text("a\n1\n") @@ -124,6 +125,7 @@ def test_read_json_invalid_json(tmp_path, monkeypatch): with pytest.raises(ValueError): read_json(source) + def test_read_http_csv(monkeypatch): response = Mock() response.text = "id,name\n1,bob\n2,alice" @@ -134,13 +136,12 @@ def test_read_http_csv(monkeypatch): lambda url, timeout: response, ) - df = read_http_csv({ - "url": "https://example.com/test.csv" - }) + df = read_http_csv({"url": "https://example.com/test.csv"}) assert len(df) == 2 assert list(df.columns) == ["id", "name"] + def test_read_http_csv_multiple_urls(monkeypatch): response = Mock() @@ -152,12 +153,9 @@ def test_read_http_csv_multiple_urls(monkeypatch): lambda url, timeout: response, ) - df = read_http_csv({ - "urls": [ - "https://example.com/a.csv", - "https://example.com/b.csv" - ] - }) + df = read_http_csv( + {"urls": ["https://example.com/a.csv", "https://example.com/b.csv"]} + ) assert len(df) == 2 @@ -165,22 +163,22 @@ def test_read_http_csv_connection_error(monkeypatch): def fake_get(url, timeout): raise requests.RequestException("network failure") - + monkeypatch.setattr( "readers.readers.requests.get", fake_get, ) with pytest.raises(ConnectionError): - read_http_csv({ - "url": "https://example.com/test.csv" - }) + read_http_csv({"url": "https://example.com/test.csv"}) + def test_read_http_csv_missing_url(): with pytest.raises(ValueError): read_http_csv({}) + def test_read_http_json(monkeypatch): response = Mock() @@ -206,33 +204,35 @@ def test_read_http_json(monkeypatch): lambda url, timeout: response, ) - df = read_http_json({ - "url": "https://api.open-meteo.com", - "json_root": "hourly", - }) + df = read_http_json( + { + "url": "https://api.open-meteo.com", + "json_root": "hourly", + } + ) assert len(df) == 2 assert "latitude" in df.columns assert "longitude" in df.columns assert "temperature_2m" in df.columns + def test_read_http_json_connection_error(monkeypatch): def fake_get(url, timeout): raise requests.RequestException("connection failure") - + monkeypatch.setattr( "readers.readers.requests.get", fake_get, ) with pytest.raises(ConnectionError): - read_http_json({ - "url": "https://api.open-meteo.com" - }) + read_http_json({"url": "https://api.open-meteo.com"}) + def test_read_http_json_invalid_json(monkeypatch): - + response = Mock() response.raise_for_status.return_value = None @@ -244,9 +244,8 @@ def test_read_http_json_invalid_json(monkeypatch): ) with pytest.raises(ValueError): - read_http_json({ - "url": "https://api.open-meto.com" - }) + read_http_json({"url": "https://api.open-meto.com"}) + def test_read_http_json_invalid_root(monkeypatch): @@ -264,7 +263,9 @@ def test_read_http_json_invalid_root(monkeypatch): ) with pytest.raises(ValueError): - read_http_json({ - "url": "https://api.open-meteo.com", - "json_root": "hourly", - }) \ No newline at end of file + read_http_json( + { + "url": "https://api.open-meteo.com", + "json_root": "hourly", + } + ) diff --git a/src/tests/test_resolver.py b/src/tests/test_resolver.py index 478af1e..17d42b5 100644 --- a/src/tests/test_resolver.py +++ b/src/tests/test_resolver.py @@ -1,9 +1,6 @@ import pytest -from sources.resolver import ( - resolve_noaa_gsom, - resolve_open_meteo, - resolve_source -) +from sources.resolver import resolve_noaa_gsom, resolve_open_meteo + def test_noaa_resolver_expands_urls(): source = { @@ -14,7 +11,7 @@ def test_noaa_resolver_expands_urls(): "target_table": "stg", "pk": ["station"], "schema": {"station": "str"}, - "rules": [] + "rules": [], } resolved = resolve_noaa_gsom(source) @@ -23,17 +20,14 @@ def test_noaa_resolver_expands_urls(): assert len(resolved["urls"]) == 2 assert resolved["urls"][0].endswith("US1.csv") + def test_noaa_missing_stations_raises(): - source = { - "name": "bad", - "type": "noaa_gsom", - "base_url": "x", - "stations": [] - } + source = {"name": "bad", "type": "noaa_gsom", "base_url": "x", "stations": []} with pytest.raises(ValueError): resolve_noaa_gsom(source) + def test_open_meteo_url_build(): source = { "name": "open_meteo", @@ -43,16 +37,14 @@ def test_open_meteo_url_build(): "longitude": 20, "start_date": "2024-01-01", "end_date": "2024-01-02", - "params": { - "hourly": ["temperature_2m"] - }, + "params": {"hourly": ["temperature_2m"]}, "target_table": "t", "pk": ["time"], "schema": {}, - "rules": [] + "rules": [], } resolved = resolve_open_meteo(source) assert "temperature_2m" in resolved["url"] - assert "latitude=10" in resolved["url"] \ No newline at end of file + assert "latitude=10" in resolved["url"] diff --git a/src/tests/test_validator.py b/src/tests/test_validator.py index 657a889..3db2325 100644 --- a/src/tests/test_validator.py +++ b/src/tests/test_validator.py @@ -55,6 +55,7 @@ def test_apply_schema_casts_invalid_type(): with pytest.raises(ValueError): apply_schema_casts(df, schema, "test_source") + def test_apply_schema_casts_missing_float_column(): df = pd.DataFrame({"station": ["ABC"]}) @@ -66,6 +67,7 @@ def test_apply_schema_casts_missing_float_column(): assert result["awnd"].isna().all() assert len(rejects) == 0 + def test_apply_schema_casts_missing_string_column(): df = pd.DataFrame({"station": ["ABC"]}) @@ -77,6 +79,7 @@ def test_apply_schema_casts_missing_string_column(): assert result["name"].isna().all() assert len(rejects) == 0 + def test_apply_schema_casts_missing_int_column(): df = pd.DataFrame({"station": ["ABC"]}) @@ -88,6 +91,7 @@ def test_apply_schema_casts_missing_int_column(): assert result["elevation"].isna().all() assert len(rejects) == 0 + def test_apply_schema_casts_missing_datetime_column(): df = pd.DataFrame({"station": ["ABC"]}) @@ -99,6 +103,7 @@ def test_apply_schema_casts_missing_datetime_column(): assert result["date"].isna().all() assert len(rejects) == 0 + def test_apply_schema_casts_invalid_float_creates_reject(): df = pd.DataFrame({"prcp": ["1.5", "abc"]}) @@ -110,6 +115,7 @@ def test_apply_schema_casts_invalid_float_creates_reject(): assert len(rejects) == 1 assert rejects[0]["reason"] == "schema_cast_failed:prcp" + def test_enforce_required_valid(): df = pd.DataFrame({"station": ["ABC"]}) diff --git a/src/validators/validators.py b/src/validators/validators.py index aa33011..f8402b0 100644 --- a/src/validators/validators.py +++ b/src/validators/validators.py @@ -45,7 +45,7 @@ def apply_schema_casts(df, schema, source_name): elif col_type == "str": try: converted = df[col].astype(col_type) - except Exception as e: + except Exception: raise ValueError(f"Failed to cast column {col} to {target_type}") else: try: