feat: add source resolution framework and external source ingestion support

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]
 

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

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

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

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",
+    }