Add output parameter to BCB modules for raw data retrieval

Author: wilsonfreitas

.claude/plans/text-output.md

@@ -0,0 +1,149 @@
+# Plan: Add `output="text"` to All BCB Modules
+
+## Context
+
+The package's primary purpose is fetching BCB API data into pandas DataFrames. Users building SOR/SOT/SPEC data pipelines need to persist raw downloaded data before transformation — it's bad practice to serialize a DataFrame back to text (lossy, format-dependent). Each BCB module already holds the raw text internally but doesn't expose it. This plan adds `output="text"` to all public `get()` / `collect()` functions so pipelines can save the exact bytes returned by BCB before any pandas processing.
+
+---
+
+## Scope
+
+All three modules: **OData**, **SGS**, **Currency**
+
+---
+
+## Interface
+
+Add `output: str = "dataframe"` parameter to:
+- `EndpointQuery.collect(output=...)` → `"dataframe"` returns `pd.DataFrame`, `"text"` returns `str`
+- `Endpoint.get(*args, output=..., **kwargs)` → same
+- `sgs.get(codes, ..., output=...)` → `"dataframe"` unchanged, `"text"` returns `str` (single code) or `dict[int, str]` (multiple codes, keyed by code value)
+- `currency.get(symbols, ..., output=...)` → `"dataframe"` unchanged, `"text"` returns `str` (single symbol) or `dict[str, str]` (multiple symbols, keyed by ISO symbol)
+
+Use `@overload` + `Literal["text", "dataframe"]` for mypy --strict compliance on each function.
+
+---
+
+## What "text" contains per module
+
+| Module | Raw text format | Source |
+|--------|----------------|--------|
+| OData  | OData JSON response: `{"@odata.context": "...", "value": [...]}` | `ODataQuery.text()` (already exists in `framework.py:505`) |
+| SGS    | BCB SGS JSON array: `[{"data": "01/01/2024", "valor": "100.5"}, ...]` | `sgs.get_json()` (already exists in `sgs/__init__.py:160`) |
+| Currency | BCB PTAX semicolon-delimited CSV | `res.text` inside `_get_symbol()` |
+
+---
+
+## File Changes
+
+### 1. `bcb/odata/api.py`
+
+**`EndpointQuery.collect()`** — add `output` param:
+```python
+def collect(self, output: str = "dataframe") -> Union[pd.DataFrame, str]:
+    if output == "text":
+        return self.text()  # inherited from ODataQuery in framework.py:505
+    # ... existing DataFrame logic unchanged
+```
+
+**`Endpoint.get()`** — intercept `output` kwarg before it reaches `_query.parameters()`:
+```python
+output_format = "dataframe"
+for k, val in kwargs.items():
+    if k == "limit": ...
+    elif k == "output":
+        output_format = val
+    else:
+        _query.parameters(**{k: val})
+...
+data = _query.collect(output=output_format)
+```
+
+Add `@overload` stubs and update return type to `Union[pd.DataFrame, str]`.
+
+### 2. `bcb/sgs/__init__.py`
+
+**`sgs.get()`** — add `output` param with early-return branch:
+```python
+def get(codes, start, end, last, multi, freq, output="dataframe"):
+    if output == "text":
+        results = {c.value: get_json(c.value, start, end, last) for c in _codes(codes)}
+        # single code → str, multiple codes → dict[int, str]
+        values = list(results.values())
+        return values[0] if len(values) == 1 else results
+    # ... existing DataFrame logic unchanged
+```
+
+Add `@overload` stubs:
+- `output: Literal["dataframe"]` → `Union[pd.DataFrame, List[pd.DataFrame]]`
+- `output: Literal["text"]` → `Union[str, dict[int, str]]`
+
+### 3. `bcb/currency.py`
+
+**New helper `_fetch_symbol_response()`**: extracts shared HTTP logic from `_get_symbol()` to avoid duplication.
+
+```python
+def _fetch_symbol_response(symbol, start_date, end_date) -> Optional[httpx.Response]:
+    try:
+        cid = _get_currency_id(symbol)
+    except CurrencyNotFoundError:
+        return None
+    res = httpx.get(_currency_url(cid, start_date, end_date), follow_redirects=True)
+    if res.headers["Content-Type"].startswith("text/html"):
+        # existing HTML error warn logic (moved from _get_symbol)
+        return None
+    return res
+
+def _get_symbol(symbol, start_date, end_date) -> Optional[pd.DataFrame]:
+    res = _fetch_symbol_response(symbol, start_date, end_date)
+    if res is None:
+        return None
+    # ... existing CSV parse logic (unchanged)
+
+def _get_symbol_text(symbol, start_date, end_date) -> Optional[str]:
+    res = _fetch_symbol_response(symbol, start_date, end_date)
+    return res.text if res is not None else None
+```
+
+**`currency.get()`** — add `output` param:
+```python
+if output == "text":
+    results = {s: _get_symbol_text(s, start, end) for s in symbols}
+    results = {k: v for k, v in results.items() if v is not None}
+    if not results:
+        raise CurrencyNotFoundError(...)
+    return results[symbols[0]] if len(symbols) == 1 else results
+```
+
+Add `@overload` stubs:
+- `output: Literal["dataframe"]` → `pd.DataFrame`
+- `output: Literal["text"]` → `Union[str, dict[str, str]]`
+
+---
+
+## Tests
+
+Add to existing test files (using mocked HTTP via `pytest-httpx`):
+
+- **`tests/test_odata.py`**: `EndpointQuery.collect(output="text")` returns str; `Endpoint.get(output="text")` returns str.
+- **`tests/test_sgs.py`**: `sgs.get(1, ..., output="text")` returns `str`; `sgs.get([1, 2], ..., output="text")` returns `dict[int, str]`.
+- **`tests/test_currency.py`**: `currency.get("USD", ..., output="text")` returns `str`; `currency.get(["USD", "EUR"], ..., output="text")` returns `dict[str, str]`.
+
+---
+
+## Verification
+
+```bash
+# Unit tests (mocked)
+poetry run pytest -m "not integration" tests/test_odata.py tests/test_sgs.py tests/test_currency.py
+
+# Type check
+poetry run mypy bcb/
+
+# Quick smoke test (live)
+poetry run python -c "
+from bcb import sgs
+text = sgs.get(1, last=3, output='text')
+print(type(text), text[:80])
+"
+```

bcb/currency.py

@@ -2,7 +2,7 @@
 import warnings
 from datetime import date, timedelta
 from io import BytesIO, StringIO
-from typing import List, Optional, Union
+from typing import Dict, List, Literal, Optional, Union, overload
 
 import httpx
 import numpy as np
@@ -121,16 +121,15 @@ def _get_currency_id(symbol: str) -> int:
     return int(matches.max())
 
 
-def _get_symbol(
+def _fetch_symbol_response(
     symbol: str, start_date: DateInput, end_date: DateInput
-) -> Optional[pd.DataFrame]:
+) -> Optional[httpx.Response]:
     try:
         cid = _get_currency_id(symbol)
     except CurrencyNotFoundError:
         return None
     url = _currency_url(cid, start_date, end_date)
     res = httpx.get(url, follow_redirects=True)
-
     if res.headers["Content-Type"].startswith("text/html"):
         doc = html.parse(BytesIO(res.content)).getroot()
         xpath = "//div[@class='msgErro']"
@@ -141,7 +140,15 @@ def _get_symbol(
         msg = f"BCB API returned error: {x} - {symbol}"
         warnings.warn(msg)
         return None
+    return res
+
 
+def _get_symbol(
+    symbol: str, start_date: DateInput, end_date: DateInput
+) -> Optional[pd.DataFrame]:
+    res = _fetch_symbol_response(symbol, start_date, end_date)
+    if res is None:
+        return None
     columns = ["Date", "aa", "bb", "cc", "bid", "ask", "dd", "ee"]
     df = pd.read_csv(
         StringIO(res.text), delimiter=";", header=None, names=columns, dtype=str
@@ -159,13 +166,43 @@ def _get_symbol(
     return df1
 
 
+def _get_symbol_text(
+    symbol: str, start_date: DateInput, end_date: DateInput
+) -> Optional[str]:
+    res = _fetch_symbol_response(symbol, start_date, end_date)
+    return res.text if res is not None else None
+
+
+@overload
+def get(
+    symbols: Union[str, List[str]],
+    start: DateInput,
+    end: DateInput,
+    side: str = ...,
+    groupby: str = ...,
+    output: Literal["dataframe"] = ...,
+) -> pd.DataFrame: ...
+
+
+@overload
+def get(
+    symbols: Union[str, List[str]],
+    start: DateInput,
+    end: DateInput,
+    side: str = ...,
+    groupby: str = ...,
+    output: Literal["text"] = ...,
+) -> Union[str, Dict[str, str]]: ...
+
+
 def get(
     symbols: Union[str, List[str]],
     start: DateInput,
     end: DateInput,
     side: str = "ask",
     groupby: str = "symbol",
-) -> pd.DataFrame:
+    output: str = "dataframe",
+) -> Union[pd.DataFrame, str, Dict[str, str]]:
     """
     Retorna um DataFrame pandas com séries temporais com taxas de câmbio.
 
@@ -204,6 +241,19 @@ def get(
     """
     if isinstance(symbols, str):
         symbols = [symbols]
+
+    if output == "text":
+        results: Dict[str, str] = {}
+        for symbol in symbols:
+            raw = _get_symbol_text(symbol, start, end)
+            if raw is not None:
+                results[symbol] = raw
+        if not results:
+            raise CurrencyNotFoundError(f"Currency not found: {symbols}")
+        if len(symbols) == 1:
+            return results[symbols[0]]
+        return results
+
     dss = []
     for symbol in symbols:
         df1 = _get_symbol(symbol, start, end)
@@ -219,6 +269,8 @@ def get(
                 return df
             elif groupby == "side":
                 return df.reorder_levels([1, 0], axis=1).sort_index(axis=1)
+            else:
+                raise ValueError("Unknown groupby value, use: symbol, side")
         else:
             raise ValueError("Unknown side value, use: bid, ask, both")
     else:

bcb/odata/api.py

@@ -1,4 +1,4 @@
-from typing import Any, Optional
+from typing import Any, Literal, Optional, Union, overload
 from .framework import (
     ODataEntitySet,
     ODataFunctionImport,
@@ -50,7 +50,15 @@ def __init__(
         super().__init__(entity, url)
         self._date_columns: list[str] = date_columns or []
 
-    def collect(self) -> pd.DataFrame:
+    @overload
+    def collect(self, output: Literal["dataframe"] = ...) -> pd.DataFrame: ...
+
+    @overload
+    def collect(self, output: Literal["text"]) -> str: ...
+
+    def collect(self, output: str = "dataframe") -> Union[pd.DataFrame, str]:
+        if output == "text":
+            return self.text()
         raw_data = super().collect()
         data = pd.DataFrame(raw_data["value"])
         if not self._raw:
@@ -109,19 +117,21 @@ def __init__(
         self._url = url
         self._date_columns: list[str] = date_columns or []
 
-    def get(self, *args: Any, **kwargs: Any) -> pd.DataFrame:
+    def get(self, *args: Any, **kwargs: Any) -> Union[pd.DataFrame, str]:
         """
         Executa a consulta na API OData e retorna o resultado.
 
         Parameters
         ----------
         *args : argumentos para a consulta
 
-        **kwargs : argumentos para a consulta
+        **kwargs : argumentos para a consulta. Use ``output='text'`` to get
+            the raw OData JSON response string instead of a DataFrame.
 
         Returns
         -------
-        pd.DataFrame: resultado da consulta
+        pd.DataFrame or str: resultado da consulta. Returns a DataFrame by
+            default; returns a raw JSON string when ``output='text'``.
         """
         _query = EndpointQuery(self._entity, self._url, self._date_columns)
         for arg in args:
@@ -132,20 +142,26 @@ def get(self, *args: Any, **kwargs: Any) -> pd.DataFrame:
             elif isinstance(arg, ODataProperty):
                 _query.select(arg)
         verbose = False
+        output_format = "dataframe"
         for k, val in kwargs.items():
             if k == "limit":
                 _query.limit(val)
             elif k == "skip":
                 _query.skip(val)
             elif k == "verbose":
                 verbose = val
+            elif k == "output":
+                output_format = val
             else:
                 _query.parameters(**{k: val})
         _query.format("application/json")
 
         if verbose:
             _query.show()
-        data = _query.collect()
+        if output_format == "text":
+            data = _query.collect(output="text")
+        else:
+            data = _query.collect()
         _query.reset()
         return data