diff --git a/config/sources.yml b/config/sources.yml index 9e5d1f2..09c6040 100644 --- a/config/sources.yml +++ b/config/sources.yml @@ -4,9 +4,101 @@ defaults: on_conflict: upsert sources: + - name: noaa_test + 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] + + schema: + station: str + name: str + date: datetime + prcp: float + snow: float + tmax: float + tmin: float + awnd: float + + rules: + - prcp >= 0 + - snow >= 0 + - tmax >= tmin + + time_filter: + column: date + 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 + 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 201f02d..d7ac3c7 100644 --- a/src/cleaners/cleaners.py +++ b/src/cleaners/cleaners.py @@ -30,3 +30,32 @@ 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 diff --git a/src/main.py b/src/main.py index aa9e37b..dc6bff5 100644 --- a/src/main.py +++ b/src/main.py @@ -3,8 +3,14 @@ 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 +33,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 diff --git a/src/readers/readers.py b/src/readers/readers.py index ad33063..356a869 100644 --- a/src/readers/readers.py +++ b/src/readers/readers.py @@ -1,9 +1,11 @@ import json -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 +18,10 @@ 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) + elif source["type"] == "http_json": + return read_http_json(source) else: raise ValueError(f"Unsupported source type {source['type']}") @@ -92,3 +98,99 @@ 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"] + + 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: + 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: + 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 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 new file mode 100644 index 0000000..1ece30f --- /dev/null +++ b/src/sources/resolver.py @@ -0,0 +1,97 @@ +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) + elif source_type == "open_meteo": + return resolve_open_meteo(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 isinstance(base_url, str) or not base_url.strip(): + 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 = [] + + 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 = { + "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 + + +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", + } diff --git a/src/tests/test_reader.py b/src/tests/test_reader.py index 4673dfc..8e73706 100644 --- a/src/tests/test_reader.py +++ b/src/tests/test_reader.py @@ -1,8 +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): @@ -115,3 +124,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", + } + ) diff --git a/src/tests/test_resolver.py b/src/tests/test_resolver.py new file mode 100644 index 0000000..17d42b5 --- /dev/null +++ b/src/tests/test_resolver.py @@ -0,0 +1,50 @@ +import pytest +from sources.resolver import resolve_noaa_gsom, resolve_open_meteo + + +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"] diff --git a/src/tests/test_validator.py b/src/tests/test_validator.py index d0d6912..3db2325 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(): @@ -53,6 +56,66 @@ def test_apply_schema_casts_invalid_type(): 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"]}) diff --git a/src/validators/validators.py b/src/validators/validators.py index 88153ee..f8402b0 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: + 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