reformatting and integration of feedback from PR review

Author: jbrand-dsp

docs/supported-sources/frankfurter.md

@@ -106,7 +106,7 @@ ingestr ingest \
 - **Primary Key**: Composite key of `date` and `currency_name`.
 - **Notes**:
   - An optional start and end date can be added via the arguments `--interval-start` and optionally `--interval-end` to define the date range (see examples below). If no start date is specified, the date will default today's date (and thus return the latest exchange rates).
-  - If a start date but no end date is specified, then the end date will default to the start date and ingestr will retrieve data for the specified start date only.
+  - If a start date but no end date is specified, then the end date will default to today's date and ingestr will retrieve data up until the latest published data.
   - Note that the [Frankfurter API](https://www.frankfurter.dev/) only publishes updates Monday-Friday. If the given date is on the weekend, the date will default to the previous Friday. The source is however implemented in ingestr in such a way as to avoid duplicating rows of data in this case (see [Incremental Loading - Replace](https://bruin-data.github.io/ingestr/getting-started/incremental-loading.html)).
 
 #### **Example Table: Handling Weekend Dates**

ingestr/main_test.py

@@ -3030,23 +3030,26 @@ def exchange_rate_on_specific_date(dest_uri):
         schema = f"testschema_frankfurter_{get_random_string(5)}"
         dest_table = f"{schema}.frankfurter_{get_random_string(5)}"
         start_date = "2025-01-03"
+        end_date = "2025-01-03"
         result = invoke_ingest_command(
             "frankfurter://",
             "exchange_rates",
             dest_uri,
             dest_table,
             interval_start=start_date,
+            interval_end=end_date,
         )
         assert result.exit_code == 0, f"Ingestion failed: {result.output}"
 
         dest_engine = sqlalchemy.create_engine(dest_uri)
-        query = f"SELECT rate FROM {dest_table} WHERE currency_name = 'GBP'"
+    
+        query = f"SELECT rate FROM {dest_table} WHERE currency_code = 'GBP'"
         with dest_engine.connect() as conn:
             rows = conn.execute(query).fetchall()
 
         # Assert that the rate for GBP is 0.82993
         assert len(rows) > 0, "No data found for GBP"
-        assert rows[0][0] == 0.82993, f"Expected rate 0.82993, but got {rows[0][0]}"
+        assert abs(rows[0][0] - 0.82993) <= 1e-6, f"Expected rate 0.82993, but got {rows[0][0]}"
 
     return [
         invalid_source_table,

ingestr/src/frankfurter/__init__.py

@@ -1,4 +1,4 @@
-from typing import Any, Iterator, Optional
+from typing import Any, Iterator
 
 import dlt
 from dlt.common.pendulum import pendulum
@@ -13,25 +13,26 @@
     max_table_nesting=0,
 )
 def frankfurter_source(
-    table: str,
-    start_date: Optional[TAnyDateTime] = None,
-    end_date: Optional[TAnyDateTime] = None,
+    start_date: TAnyDateTime,
+    end_date: TAnyDateTime,
 ) -> Any:
     """
     A dlt source for the frankfurter.dev API. It groups several resources (in this case frankfurter.dev API endpoints) containing
     various types of data: currencies, latest rates, historical rates.
-
-    Returns the appropriate resource based on the provided parameters.
     """
-    # Determine which resource to return based on the `table` parameter
-    if table == "currencies":
-        return currencies()
-
-    elif table == "latest":
-        return latest()
+    dateTime = dlt.sources.incremental(
+        "date",
+        initial_value=start_date,
+        end_value=end_date,
+        range_start="closed",
+        range_end="closed",
+    )
 
-    elif table == "exchange_rates":
-        return exchange_rates(start_date=start_date, end_date=end_date)
+    return (
+        currencies(),
+        latest(),
+        exchange_rates(start_date=dateTime, end_date=end_date),
+    )
 
 
 @dlt.resource(
@@ -53,7 +54,7 @@ def currencies() -> Iterator[dict]:
 
 
 @dlt.resource(
-    write_disposition="replace",
+    write_disposition="merge",
     columns={
         "date": {"data_type": "text"},
         "currency_code": {"data_type": "text"},
@@ -69,13 +70,12 @@ def latest() -> Iterator[dict]:
     url = "latest?"
 
     # Fetch data
-    latest_data = get_path_with_retry(url)
+    data = get_path_with_retry(url)
 
     # Extract rates and base currency
-    rates = latest_data["rates"]
+    rates = data["rates"]
 
-    # Prepare the date
-    date = pendulum.now("Europe/Berlin").to_date_string()
+    date = pendulum.parse(data["date"])
 
     # Add the base currency (EUR) with a rate of 1.0
     yield {
@@ -94,25 +94,34 @@ def latest() -> Iterator[dict]:
 
 
 @dlt.resource(
-    write_disposition="replace",
+    write_disposition="merge",
     columns={
         "date": {"data_type": "text"},
         "currency_code": {"data_type": "text"},
         "rate": {"data_type": "double"},
     },
-    primary_key=["date", "currency_code"],  # Composite primary key
+    primary_key=("date", "currency_code"),  # Composite primary key
 )
 def exchange_rates(
-    start_date: TAnyDateTime,
     end_date: TAnyDateTime,
+    start_date: dlt.sources.incremental[TAnyDateTime] = dlt.sources.incremental("date"),
 ) -> Iterator[dict]:
     """
     Fetches exchange rates for a specified date range.
-    If only start_date is provided, fetches data for that date.
+    If only start_date is provided, fetches data until now.
     If both start_date and end_date are provided, fetches data for each day in the range.
     """
-    start_date_str = ensure_pendulum_datetime(start_date).format("YYYY-MM-DD")
-    end_date_str = ensure_pendulum_datetime(end_date).format("YYYY-MM-DD")
+    # Ensure start_date.last_value is not None
+    if start_date.last_value is None:
+        raise ValueError("start_date.last_value cannot be None")
+
+    # Ensure start_date.last_value is a pendulum.DateTime object
+    start_date_obj = ensure_pendulum_datetime(start_date.last_value)
+    start_date_str = start_date_obj.format("YYYY-MM-DD")
+
+    # Ensure end_date is a pendulum.DateTime object
+    end_date_obj = ensure_pendulum_datetime(end_date)
+    end_date_str = end_date_obj.format("YYYY-MM-DD")
 
     # Compose the URL
     url = f"{start_date_str}..{end_date_str}?"
@@ -121,22 +130,23 @@ def exchange_rates(
     data = get_path_with_retry(url)
 
     # Extract base currency and rates from the API response
-    base_currency = data["base"]
     rates = data["rates"]
 
     # Iterate over the rates dictionary (one entry per date)
     for date, daily_rates in rates.items():
+        formatted_date = pendulum.parse(date)
+
         # Add the base currency with a rate of 1.0
         yield {
-            "date": date,
-            "currency_code": base_currency,
+            "date": formatted_date,
+            "currency_code": "EUR",
             "rate": 1.0,
         }
 
         # Add all other currencies and their rates
         for currency_code, rate in daily_rates.items():
             yield {
-                "date": date,
+                "date": formatted_date,
                 "currency_code": currency_code,
                 "rate": rate,
             }

ingestr/src/frankfurter/helpers.py

@@ -8,7 +8,7 @@
 
 
 def get_url_with_retry(url: str) -> StrAny:
-    r = requests.get(url)
+    r = requests.get(url, timeout=5)
     return r.json()  # type: ignore
 
 

ingestr/src/sources.py

@@ -67,8 +67,6 @@
 from ingestr.src.facebook_ads import facebook_ads_source, facebook_insights_source
 from ingestr.src.filesystem import readers
 from ingestr.src.filters import table_adapter_exclude_columns
-from ingestr.src.frankfurter import frankfurter_source
-from ingestr.src.frankfurter.helpers import validate_dates
 from ingestr.src.github import github_reactions, github_repo_events, github_stargazers
 from ingestr.src.google_ads import google_ads
 from ingestr.src.google_analytics import google_analytics
@@ -2050,31 +2048,32 @@ def handles_incrementality(self) -> bool:
         return True
 
     def dlt_source(self, uri: str, table: str, **kwargs):
-        # start and end dates only assigned and validated for exchange_rates table
-        # Note: if an end date but no start date is provided, start date and end date will be set to current date
-        if table == "exchange_rates":
-            if kwargs.get("interval_start"):
-                start_date = ensure_pendulum_datetime(str(kwargs.get("interval_start")))
-                if kwargs.get("interval_end"):
-                    end_date = ensure_pendulum_datetime(str(kwargs.get("interval_end")))
-                else:
-                    end_date = start_date
+        if kwargs.get("incremental_key"):
+            raise ValueError(
+                "Frankfurter takes care of incrementality on its own, you should not provide incremental_key"
+            )
+
+        if kwargs.get("interval_start"):
+            start_date = ensure_pendulum_datetime(str(kwargs.get("interval_start")))
+            if kwargs.get("interval_end"):
+                end_date = ensure_pendulum_datetime(str(kwargs.get("interval_end")))
             else:
-                start_date = pendulum.now("Europe/Berlin")
                 end_date = pendulum.now("Europe/Berlin")
-            validate_dates(start_date=start_date, end_date=end_date)
-
-        # For currencies and latest tables, set start and end dates to current date
         else:
             start_date = pendulum.now("Europe/Berlin")
             end_date = pendulum.now("Europe/Berlin")
 
-        # Validate table
-        if table not in ["currencies", "latest", "exchange_rates"]:
-            raise UnsupportedResourceError(table, "Frankfurter")
+        from ingestr.src.frankfurter import frankfurter_source
+        from ingestr.src.frankfurter.helpers import validate_dates
+
+        validate_dates(start_date=start_date, end_date=end_date)
 
-        return frankfurter_source(
-            table=table,
+        src = frankfurter_source(
             start_date=start_date,
             end_date=end_date,
         )
+
+        if table not in src.resources:
+            raise UnsupportedResourceError(table, "Frankfurter")
+
+        return src.with_resources(table)