From 10b62bda8c5bf3f5c2614370afd15bd0af780317 Mon Sep 17 00:00:00 2001 From: Wilson Freitas Date: Sun, 14 Jun 2026 20:35:40 -0300 Subject: [PATCH] Add per-request timeout support --- README.md | 22 ++++ bcb/currency.py | 149 +++++++++++++++++++------- bcb/http.py | 11 +- bcb/odata/api.py | 83 ++++++++++----- bcb/odata/framework.py | 60 ++++++++--- bcb/sgs/__init__.py | 50 +++++++-- docs/currency.rst | 12 +++ docs/odata.rst | 23 ++++ docs/sgs.rst | 19 ++++ tests/test_request_timeout.py | 192 ++++++++++++++++++++++++++++++++++ 10 files changed, 533 insertions(+), 88 deletions(-) create mode 100644 tests/test_request_timeout.py diff --git a/README.md b/README.md index bd6cb93..77abfc7 100644 --- a/README.md +++ b/README.md @@ -124,6 +124,28 @@ async def main(): asyncio.run(main()) ``` +### P: Como aumento o timeout de uma consulta lenta? +**R:** Passe ``timeout`` na própria chamada. O valor é local daquela chamada e +não altera o cliente HTTP compartilhado globalmente: + +```python +from bcb import sgs, currency, Expectativas + +# SGS +selic = sgs.get(11, start="1990-01-01", timeout=120) + +# Currency +usd = currency.get("USD", start="1980-01-01", end="2026-01-01", timeout=120) + +# OData +em = Expectativas(timeout=120) +ep = em.get_endpoint("ExpectativasMercadoAnuais") +df = ep.query().limit(1000).collect(timeout=120) +``` + +O padrão continua sendo 30 segundos. No SGS, o timeout vale por tentativa HTTP; +quando houver retry, cada tentativa usa o mesmo valor. + ### P: Como trato erros e dados faltantes? **R:** A biblioteca lança exceções específicas: - `CurrencyNotFoundError`: Símbolo de moeda não encontrado diff --git a/bcb/currency.py b/bcb/currency.py index 8be5cf6..cefa408 100644 --- a/bcb/currency.py +++ b/bcb/currency.py @@ -15,10 +15,12 @@ from lxml import etree, html from bcb.http import ( + RequestTimeout, get_async_client, get_client, raise_for_request_error, raise_for_status, + timeout_kwargs, ) from bcb.exceptions import BCBAPIError, CurrencyNotFoundError from bcb.utils import Date, DateInput @@ -130,6 +132,8 @@ def clear_cache(cache: _ThreadSafeCache | None = None) -> None: def _currency_id_list( cache: _ThreadSafeCache | None = None, + *, + timeout: RequestTimeout = None, ) -> pd.DataFrame: """Fetch list of available currency IDs and names. @@ -164,7 +168,7 @@ def _currency_id_list( ) logger.debug(f"Fetching currency ID list from {url1}") try: - res = get_client().get(url1) + res = get_client().get(url1, **timeout_kwargs(timeout)) except httpx.HTTPError as ex: raise_for_request_error(ex, context="Currency ID list") logger.debug( @@ -186,7 +190,11 @@ def _currency_id_list( def _get_valid_currency_list( - _date: date, n: int = 0, max_rollback: int = 30 + _date: date, + n: int = 0, + max_rollback: int = 30, + *, + timeout: RequestTimeout = None, ) -> "httpx.Response": """Fetch currency list CSV, rolling back dates if necessary. @@ -225,7 +233,7 @@ def _get_valid_currency_list( url2 = f"https://www4.bcb.gov.br/Download/fechamento/M{_date:%Y%m%d}.csv" logger.debug(f"Fetching currency list from {url2}") try: - res = get_client().get(url2) + res = get_client().get(url2, **timeout_kwargs(timeout)) except httpx.HTTPError as ex: # Connection error: retry same date up to 3 times if n >= 3: @@ -233,7 +241,7 @@ def _get_valid_currency_list( logger.warning( f"Connection error fetching {url2}, retrying (attempt {n + 1}/3)" ) - return _get_valid_currency_list(_date, n + 1, max_rollback) + return _get_valid_currency_list(_date, n + 1, max_rollback, timeout=timeout) logger.debug( f"Currency list response: status={res.status_code}, length={len(res.content)}" @@ -245,11 +253,15 @@ def _get_valid_currency_list( # Non-200 response (file not found for date): roll back to previous day logger.debug(f"Currency list not found for {_date}, rolling back to previous day") - return _get_valid_currency_list(_date - timedelta(1), 0, max_rollback) + return _get_valid_currency_list( + _date - timedelta(1), 0, max_rollback, timeout=timeout + ) def get_currency_list( cache: _ThreadSafeCache | None = None, + *, + timeout: RequestTimeout = None, ) -> pd.DataFrame: """Listagem com todas as moedas disponíveis na API e suas configurações de paridade. @@ -274,7 +286,7 @@ def get_currency_list( if cached is not None: return cached - res = _get_valid_currency_list(date.today()) + res = _get_valid_currency_list(date.today(), timeout=timeout) df = pd.read_csv(StringIO(res.text), delimiter=";") df.columns = [ "code", @@ -294,9 +306,9 @@ def get_currency_list( return df -def _get_currency_id(symbol: str) -> int: - id_list = _currency_id_list() - all_currencies = get_currency_list() +def _get_currency_id(symbol: str, *, timeout: RequestTimeout = None) -> int: + id_list = _currency_id_list(timeout=timeout) + all_currencies = get_currency_list(timeout=timeout) x = pd.merge(id_list, all_currencies, on=["name"]) matches = x.loc[x["symbol"] == symbol, "id"] if matches.empty: @@ -352,7 +364,11 @@ def _raise_currency_html_error(response: httpx.Response, symbol: str) -> NoRetur def _fetch_symbol_response( - symbol: str, start_date: DateInput, end_date: DateInput + symbol: str, + start_date: DateInput, + end_date: DateInput, + *, + timeout: RequestTimeout = None, ) -> "httpx.Response": """Fetch exchange rate CSV response for a symbol. @@ -379,11 +395,13 @@ def _fetch_symbol_response( BCBAPIError If API returns other error status codes or HTML error page """ - cid = _get_currency_id(symbol) # Raises CurrencyNotFoundError if not found + cid = _get_currency_id( + symbol, timeout=timeout + ) # Raises CurrencyNotFoundError if not found url = _currency_url(cid, start_date, end_date) logger.debug(f"Fetching currency data for {symbol} from {url.split('?')[0]}") try: - res = get_client().get(url) + res = get_client().get(url, **timeout_kwargs(timeout)) except httpx.HTTPError as ex: raise_for_request_error(ex, context=f"Currency data for {symbol}") logger.debug( @@ -496,7 +514,11 @@ def _parse_currency_types(df: pd.DataFrame) -> pd.DataFrame: def _get_symbol( - symbol: str, start_date: DateInput, end_date: DateInput + symbol: str, + start_date: DateInput, + end_date: DateInput, + *, + timeout: RequestTimeout = None, ) -> pd.DataFrame: """Fetch and parse exchange rate data for a symbol. @@ -521,7 +543,7 @@ def _get_symbol( BCBAPIError If API returns error or data format is invalid """ - res = _fetch_symbol_response(symbol, start_date, end_date) + res = _fetch_symbol_response(symbol, start_date, end_date, timeout=timeout) df = _validate_currency_csv(res.text) df = _parse_currency_dates(df) df = _parse_currency_types(df) @@ -533,7 +555,13 @@ def _get_symbol( return df1 -def _get_symbol_text(symbol: str, start_date: DateInput, end_date: DateInput) -> str: +def _get_symbol_text( + symbol: str, + start_date: DateInput, + end_date: DateInput, + *, + timeout: RequestTimeout = None, +) -> str: """Fetch exchange rate data as CSV text for a symbol. Parameters @@ -557,7 +585,7 @@ def _get_symbol_text(symbol: str, start_date: DateInput, end_date: DateInput) -> BCBAPIError If API returns error """ - res = _fetch_symbol_response(symbol, start_date, end_date) + res = _fetch_symbol_response(symbol, start_date, end_date, timeout=timeout) return res.text @@ -606,6 +634,8 @@ def get( side: CurrencySide = ..., groupby: CurrencyGroupBy = ..., output: Literal["dataframe"] = ..., + *, + timeout: RequestTimeout = ..., ) -> pd.DataFrame: ... @@ -617,6 +647,8 @@ def get( side: CurrencySide = ..., groupby: CurrencyGroupBy = ..., output: Literal["dataframe"] = ..., + *, + timeout: RequestTimeout = ..., ) -> pd.DataFrame: ... @@ -628,6 +660,8 @@ def get( side: CurrencySide = ..., groupby: CurrencyGroupBy = ..., output: Literal["text"] = ..., + *, + timeout: RequestTimeout = ..., ) -> str: ... @@ -639,6 +673,8 @@ def get( side: CurrencySide = ..., groupby: CurrencyGroupBy = ..., output: Literal["text"] = ..., + *, + timeout: RequestTimeout = ..., ) -> CurrencyTextResult: ... @@ -649,6 +685,8 @@ def get( side: CurrencySide = "ask", groupby: CurrencyGroupBy = "symbol", output: CurrencyOutput = "dataframe", + *, + timeout: RequestTimeout = None, ) -> Union[pd.DataFrame, str, Dict[str, str]]: """ Retorna um DataFrame pandas com séries temporais com taxas de câmbio. @@ -674,6 +712,9 @@ def get( por ``side``. output : {"dataframe", "text"}, default "dataframe" Define o formato de saída. Use ``"text"`` para retornar o CSV bruto. + timeout : float or httpx.Timeout, optional + Timeout por requisição HTTP, em segundos ou como ``httpx.Timeout``. + Quando omitido, usa o timeout padrão do cliente compartilhado. Returns ------- @@ -696,7 +737,7 @@ def get( results: Dict[str, str] = {} for symbol in symbols: try: - raw = _get_symbol_text(symbol, start, end) + raw = _get_symbol_text(symbol, start, end, timeout=timeout) results[symbol] = raw except CurrencyNotFoundError: pass # Skip missing currencies @@ -709,7 +750,7 @@ def get( dss = [] for symbol in symbols: try: - df1 = _get_symbol(symbol, start, end) + df1 = _get_symbol(symbol, start, end, timeout=timeout) dss.append(df1) except CurrencyNotFoundError: pass # Skip missing currencies @@ -733,6 +774,8 @@ def get( async def _async_currency_id_list( cache: _ThreadSafeCache | None = None, + *, + timeout: RequestTimeout = None, ) -> pd.DataFrame: """Async version of _currency_id_list().""" cache = cache or _DEFAULT_CACHE @@ -746,7 +789,7 @@ async def _async_currency_id_list( "method=exibeFormularioConsultaBoletim" ) try: - res = await get_async_client().get(url1) + res = await get_async_client().get(url1, **timeout_kwargs(timeout)) except httpx.HTTPError as ex: raise_for_request_error(ex, context="Currency ID list") raise_for_status( @@ -765,7 +808,11 @@ async def _async_currency_id_list( async def _async_get_valid_currency_list( - _date: date, n: int = 0, max_rollback: int = 30 + _date: date, + n: int = 0, + max_rollback: int = 30, + *, + timeout: RequestTimeout = None, ) -> "httpx.Response": """Async version of _get_valid_currency_list().""" days_rolled_back = (date.today() - _date).days @@ -777,21 +824,27 @@ async def _async_get_valid_currency_list( url2 = f"https://www4.bcb.gov.br/Download/fechamento/M{_date:%Y%m%d}.csv" try: - res = await get_async_client().get(url2) + res = await get_async_client().get(url2, **timeout_kwargs(timeout)) except httpx.HTTPError as ex: if n >= 3: raise_for_request_error(ex, context="Currency list") - return await _async_get_valid_currency_list(_date, n + 1, max_rollback) + return await _async_get_valid_currency_list( + _date, n + 1, max_rollback, timeout=timeout + ) if res.status_code == 200: return res if res.status_code == 429 or res.status_code >= 500: raise_for_status(res, context="Currency list") - return await _async_get_valid_currency_list(_date - timedelta(1), 0, max_rollback) + return await _async_get_valid_currency_list( + _date - timedelta(1), 0, max_rollback, timeout=timeout + ) async def _async_get_currency_list( cache: _ThreadSafeCache | None = None, + *, + timeout: RequestTimeout = None, ) -> pd.DataFrame: """Async version of get_currency_list().""" cache = cache or _DEFAULT_CACHE @@ -800,7 +853,7 @@ async def _async_get_currency_list( if cached is not None: return cached - res = await _async_get_valid_currency_list(date.today()) + res = await _async_get_valid_currency_list(date.today(), timeout=timeout) df = pd.read_csv(StringIO(res.text), delimiter=";") df.columns = [ "code", @@ -820,11 +873,11 @@ async def _async_get_currency_list( return df -async def _async_get_currency_id(symbol: str) -> int: +async def _async_get_currency_id(symbol: str, *, timeout: RequestTimeout = None) -> int: """Async version of _get_currency_id() with concurrent cache warming.""" id_list, all_currencies = await asyncio.gather( - _async_currency_id_list(), - _async_get_currency_list(), + _async_currency_id_list(timeout=timeout), + _async_get_currency_list(timeout=timeout), ) x = pd.merge(id_list, all_currencies, on=["name"]) matches = x.loc[x["symbol"] == symbol, "id"] @@ -834,13 +887,17 @@ async def _async_get_currency_id(symbol: str) -> int: async def _async_fetch_symbol_response( - symbol: str, start_date: DateInput, end_date: DateInput + symbol: str, + start_date: DateInput, + end_date: DateInput, + *, + timeout: RequestTimeout = None, ) -> "httpx.Response": """Async version of _fetch_symbol_response().""" - cid = await _async_get_currency_id(symbol) + cid = await _async_get_currency_id(symbol, timeout=timeout) url = _currency_url(cid, start_date, end_date) try: - res = await get_async_client().get(url) + res = await get_async_client().get(url, **timeout_kwargs(timeout)) except httpx.HTTPError as ex: raise_for_request_error(ex, context=f"Currency data for {symbol}") @@ -857,10 +914,16 @@ async def _async_fetch_symbol_response( async def _async_get_symbol( - symbol: str, start_date: DateInput, end_date: DateInput + symbol: str, + start_date: DateInput, + end_date: DateInput, + *, + timeout: RequestTimeout = None, ) -> pd.DataFrame: """Async version of _get_symbol().""" - res = await _async_fetch_symbol_response(symbol, start_date, end_date) + res = await _async_fetch_symbol_response( + symbol, start_date, end_date, timeout=timeout + ) df = _validate_currency_csv(res.text) df = _parse_currency_dates(df) df = _parse_currency_types(df) @@ -873,10 +936,16 @@ async def _async_get_symbol( async def _async_get_symbol_text( - symbol: str, start_date: DateInput, end_date: DateInput + symbol: str, + start_date: DateInput, + end_date: DateInput, + *, + timeout: RequestTimeout = None, ) -> str: """Async version of _get_symbol_text().""" - res = await _async_fetch_symbol_response(symbol, start_date, end_date) + res = await _async_fetch_symbol_response( + symbol, start_date, end_date, timeout=timeout + ) return res.text @@ -887,6 +956,8 @@ async def async_get( side: CurrencySide = "ask", groupby: CurrencyGroupBy = "symbol", output: CurrencyOutput = "dataframe", + *, + timeout: RequestTimeout = None, ) -> Union[pd.DataFrame, str, Dict[str, str]]: """ Retorna um DataFrame pandas com séries temporais com taxas de câmbio (async version). @@ -911,6 +982,9 @@ async def async_get( ``'symbol'`` ou ``'side'`` output : {"dataframe", "text"} ``'dataframe'`` ou ``'text'`` + timeout : float or httpx.Timeout, optional + Timeout por requisição HTTP, em segundos ou como ``httpx.Timeout``. + Quando omitido, usa o timeout padrão do cliente compartilhado. Returns ------- @@ -924,7 +998,10 @@ async def async_get( if output == "text": results: Dict[str, str] = {} texts = await asyncio.gather( - *[_async_get_symbol_text(symbol, start, end) for symbol in symbols], + *[ + _async_get_symbol_text(symbol, start, end, timeout=timeout) + for symbol in symbols + ], return_exceptions=True, ) for symbol, text in zip(symbols, texts, strict=True): @@ -940,7 +1017,7 @@ async def async_get( return results dss = await asyncio.gather( - *[_async_get_symbol(symbol, start, end) for symbol in symbols], + *[_async_get_symbol(symbol, start, end, timeout=timeout) for symbol in symbols], return_exceptions=True, ) valid_dss = [] diff --git a/bcb/http.py b/bcb/http.py index 0641113..9dc43a8 100644 --- a/bcb/http.py +++ b/bcb/http.py @@ -2,7 +2,7 @@ from __future__ import annotations -from typing import Callable, NoReturn, TypeVar +from typing import Any, Callable, NoReturn, TypeAlias, TypeVar import httpx from tenacity import ( @@ -21,6 +21,8 @@ # Default timeout for all HTTP requests (seconds) DEFAULT_TIMEOUT = 30.0 +RequestTimeout: TypeAlias = float | httpx.Timeout | None + # Shared synchronous HTTP client _CLIENT = httpx.Client( timeout=DEFAULT_TIMEOUT, @@ -95,6 +97,13 @@ def close_async_client() -> None: loop.create_task(aclose_async_client()) +def timeout_kwargs(timeout: RequestTimeout) -> dict[str, Any]: + """Build request kwargs without overriding the client default timeout.""" + if timeout is None: + return {} + return {"timeout": timeout} + + T = TypeVar("T") diff --git a/bcb/odata/api.py b/bcb/odata/api.py index 9e08433..e23b5d5 100644 --- a/bcb/odata/api.py +++ b/bcb/odata/api.py @@ -2,6 +2,7 @@ from typing import Any, Literal, Optional, Union, overload +from bcb.http import RequestTimeout from bcb.odata.framework import ( ODataEntitySet, ODataFunctionImport, @@ -20,8 +21,8 @@ class EndpointMeta(type): def __init__(self, *args: Any, **kwargs: Any) -> None: super().__init__(*args, **kwargs) - def __call__(self, *args: Any) -> Any: - obj = super().__call__(*args) + def __call__(self, *args: Any, **kwargs: Any) -> Any: + obj = super().__call__(*args, **kwargs) entity = args[0] if isinstance(entity, ODataEntitySet): for name, prop in entity.entity.properties.items(): @@ -49,20 +50,31 @@ def __init__( entity: Any, url: str, date_columns: Optional[list[str]] = None, + *, + timeout: RequestTimeout = None, ) -> None: - super().__init__(entity, url) + super().__init__(entity, url, timeout=timeout) self._date_columns: list[str] = date_columns or [] @overload - def collect(self, output: Literal["dataframe"] = ...) -> pd.DataFrame: ... + def collect( + self, + output: Literal["dataframe"] = ..., + *, + timeout: RequestTimeout = ..., + ) -> pd.DataFrame: ... @overload - def collect(self, output: Literal["text"]) -> str: ... + def collect( + self, output: Literal["text"], *, timeout: RequestTimeout = ... + ) -> str: ... - def collect(self, output: str = "dataframe") -> Union[pd.DataFrame, str]: + def collect( + self, output: str = "dataframe", *, timeout: RequestTimeout = None + ) -> Union[pd.DataFrame, str]: if output == "text": - return self.text() - raw_data = super().collect() + return self.text(timeout=timeout) + raw_data = super().collect(timeout=timeout) data = pd.DataFrame(raw_data["value"]) if not self._raw: if self._date_columns: @@ -85,12 +97,12 @@ def collect(self, output: str = "dataframe") -> Union[pd.DataFrame, str]: return data async def async_collect( - self, output: str = "dataframe" + self, output: str = "dataframe", *, timeout: RequestTimeout = None ) -> Union[pd.DataFrame, str]: """Async version of collect(). Awaits super().async_collect() for data fetch.""" if output == "text": - return await self.async_text() - raw_data = await super().async_collect() + return await self.async_text(timeout=timeout) + raw_data = await super().async_collect(timeout=timeout) data = pd.DataFrame(raw_data["value"]) if not self._raw: if self._date_columns: @@ -126,7 +138,12 @@ class Endpoint(metaclass=EndpointMeta): """ def __init__( - self, entity: Any, url: str, date_columns: Optional[list[str]] = None + self, + entity: Any, + url: str, + date_columns: Optional[list[str]] = None, + *, + timeout: RequestTimeout = None, ) -> None: """ Construtor da classe Endpoint. @@ -145,6 +162,7 @@ def __init__( self._entity = entity self._url = url self._date_columns: list[str] = date_columns or [] + self._timeout = timeout def get( self, @@ -155,6 +173,7 @@ def get( limit: Optional[int] = None, skip: Optional[int] = None, output: str = "dataframe", + timeout: RequestTimeout = None, verbose: bool = False, **kwargs: Any, ) -> Union[pd.DataFrame, str]: @@ -186,7 +205,9 @@ def get( 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) + _query = EndpointQuery( + self._entity, self._url, self._date_columns, timeout=self._timeout + ) # Apply explicit kwargs first if filter is not None: @@ -218,9 +239,9 @@ def get( if verbose: _query.show() if output == "text": - data = _query.collect(output="text") + data = _query.collect(output="text", timeout=timeout) else: - data = _query.collect() + data = _query.collect(timeout=timeout) _query.reset() return data @@ -232,7 +253,9 @@ def query(self) -> EndpointQuery: ------- bcb.odata.api.EndpointQuery """ - return EndpointQuery(self._entity, self._url, self._date_columns) + return EndpointQuery( + self._entity, self._url, self._date_columns, timeout=self._timeout + ) def async_query(self) -> EndpointQuery: """ @@ -243,7 +266,9 @@ def async_query(self) -> EndpointQuery: bcb.odata.api.EndpointQuery Same as query(); call async_collect() on the result """ - return EndpointQuery(self._entity, self._url, self._date_columns) + return EndpointQuery( + self._entity, self._url, self._date_columns, timeout=self._timeout + ) async def async_get( self, @@ -254,6 +279,7 @@ async def async_get( limit: Optional[int] = None, skip: Optional[int] = None, output: str = "dataframe", + timeout: RequestTimeout = None, verbose: bool = False, **kwargs: Any, ) -> Union[pd.DataFrame, str]: @@ -287,7 +313,9 @@ async def async_get( Union[pd.DataFrame, str] Resultado da consulta """ - _query = EndpointQuery(self._entity, self._url, self._date_columns) + _query = EndpointQuery( + self._entity, self._url, self._date_columns, timeout=self._timeout + ) # Apply explicit kwargs first if filter is not None: @@ -319,9 +347,9 @@ async def async_get( if verbose: _query.show() if output == "text": - data = await _query.async_collect(output="text") + data = await _query.async_collect(output="text", timeout=timeout) else: - data = await _query.async_collect() + data = await _query.async_collect(timeout=timeout) _query.reset() return data @@ -336,11 +364,12 @@ class BaseODataAPI: BASE_URL: str DATE_COLUMNS: list[str] = [] - def __init__(self) -> None: + def __init__(self, *, timeout: RequestTimeout = None) -> None: """ BaseODataAPI construtor """ - self.service = ODataService(self.BASE_URL) + self._timeout = timeout + self.service = ODataService(self.BASE_URL, timeout=timeout) def describe(self, endpoint: Optional[str] = None) -> None: """ @@ -385,7 +414,10 @@ def get_endpoint(self, endpoint: str) -> Endpoint: Se o *endpoint* fornecido é errado. """ return Endpoint( - self.service[endpoint], self.service.url, self.DATE_COLUMNS or None + self.service[endpoint], + self.service.url, + self.DATE_COLUMNS or None, + timeout=self._timeout, ) @@ -400,7 +432,7 @@ class ODataAPI(BaseODataAPI): não possuem implementação específica. """ - def __init__(self, url: str) -> None: + def __init__(self, url: str, *, timeout: RequestTimeout = None) -> None: """ Parameters ---------- @@ -417,7 +449,8 @@ def __init__(self, url: str) -> None: - Expectativas - PTAX """ - self.service = ODataService(url) + self._timeout = timeout + self.service = ODataService(url, timeout=timeout) class Expectativas(BaseODataAPI): diff --git a/bcb/odata/framework.py b/bcb/odata/framework.py index 9538a98..2a22929 100644 --- a/bcb/odata/framework.py +++ b/bcb/odata/framework.py @@ -13,10 +13,12 @@ from typing_extensions import Self from bcb.http import ( + RequestTimeout, get_async_client, get_client, raise_for_request_error, raise_for_status, + timeout_kwargs, ) from bcb.exceptions import ODataError @@ -339,9 +341,10 @@ class ODataMetadata: "edmx": "http://docs.oasis-open.org/odata/ns/edmx", } - def __init__(self, url: str) -> None: + def __init__(self, url: str, *, timeout: RequestTimeout = None) -> None: self.url = url - self._load_document() + self._timeout = timeout + self._load_document(timeout=timeout) try: _xpath = "edmx:DataServices/edm:Schema" schemas = self.doc.xpath(_xpath, namespaces=self.namespaces) @@ -361,10 +364,10 @@ def __init__(self, url: str) -> None: f"OData metadata {self.url} has invalid structure: {ex}" ) from ex - def _load_document(self) -> None: + def _load_document(self, *, timeout: RequestTimeout = None) -> None: logger.debug(f"Fetching OData metadata from {self.url}") try: - res = get_client().get(self.url) + res = get_client().get(self.url, **timeout_kwargs(timeout)) except httpx.HTTPError as ex: raise_for_request_error( ex, context=f"OData metadata {self.url}", error_cls=ODataError @@ -469,10 +472,11 @@ class ODataService: OData service root URL """ - def __init__(self, url: str) -> None: + def __init__(self, url: str, *, timeout: RequestTimeout = None) -> None: self.url = url + self._timeout = timeout try: - res = get_client().get(self.url) + res = get_client().get(self.url, **timeout_kwargs(timeout)) except httpx.HTTPError as ex: raise_for_request_error( ex, context=f"OData service {self.url}", error_cls=ODataError @@ -512,7 +516,7 @@ def __init__(self, url: str) -> None: if self._odata_context_url in _METADATA_CACHE: self.metadata = _METADATA_CACHE[self._odata_context_url] else: - self.metadata = ODataMetadata(self._odata_context_url) + self.metadata = ODataMetadata(self._odata_context_url, timeout=timeout) _METADATA_CACHE[self._odata_context_url] = self.metadata def __getitem__(self, item: str) -> Union[ODataEntitySet, ODataFunctionImport]: @@ -556,15 +560,20 @@ def describe(self) -> None: def query( self, entity_set: Union[ODataEntitySet, ODataFunctionImport] ) -> "ODataQuery": - return ODataQuery(entity_set, self.url) + return ODataQuery(entity_set, self.url, timeout=self._timeout) class ODataQuery: def __init__( - self, entity: Union[ODataEntitySet, ODataFunctionImport], url: str + self, + entity: Union[ODataEntitySet, ODataFunctionImport], + url: str, + *, + timeout: RequestTimeout = None, ) -> None: self.entity = entity self.base_url = url + self._timeout = timeout self._params: dict[str, Any] = {} self.function_parameters: dict[str, Any] = {} self._filter: list[ODataPropertyFilter] = [] @@ -649,13 +658,20 @@ def reset(self) -> None: self._orderby = [] self._params = {} - def collect(self) -> Any: + def _resolve_timeout(self, timeout: RequestTimeout) -> RequestTimeout: + if timeout is None: + return self._timeout + return timeout + + def collect(self, *, timeout: RequestTimeout = None) -> Any: url = self.odata_url() - data = _load_json_object(self.text(), context=f"OData query {url}") + data = _load_json_object( + self.text(timeout=timeout), context=f"OData query {url}" + ) _required_field(data, "value", context=f"OData query {url}") return data - async def async_text(self) -> str: + async def async_text(self, *, timeout: RequestTimeout = None) -> str: """Async version of text(). Fetches OData response using shared client.""" params = self._build_parameters() if self.is_function and len(self.function_parameters): @@ -668,7 +684,11 @@ async def async_text(self) -> str: headers = {"OData-Version": "4.0", "OData-MaxVersion": "4.0"} url = self.odata_url() try: - res = await get_async_client().get(url + "?" + qs, headers=headers) + res = await get_async_client().get( + url + "?" + qs, + headers=headers, + **timeout_kwargs(self._resolve_timeout(timeout)), + ) except httpx.HTTPError as ex: raise_for_request_error( ex, context=f"OData query {url}", error_cls=ODataError @@ -683,14 +703,16 @@ async def async_text(self) -> str: ) return res.text - async def async_collect(self) -> Any: + async def async_collect(self, *, timeout: RequestTimeout = None) -> Any: """Async version of collect(). Awaits async_text() and parses JSON.""" url = self.odata_url() - data = _load_json_object(await self.async_text(), context=f"OData query {url}") + data = _load_json_object( + await self.async_text(timeout=timeout), context=f"OData query {url}" + ) _required_field(data, "value", context=f"OData query {url}") return data - def text(self) -> str: + def text(self, *, timeout: RequestTimeout = None) -> str: params = self._build_parameters() if self.is_function and len(self.function_parameters): for p in self.entity.function.parameters: # type: ignore[union-attr] @@ -703,7 +725,11 @@ def text(self) -> str: url = self.odata_url() logger.debug(f"Fetching OData query from {url}") try: - res = get_client().get(url + "?" + qs, headers=headers) + res = get_client().get( + url + "?" + qs, + headers=headers, + **timeout_kwargs(self._resolve_timeout(timeout)), + ) except httpx.HTTPError as ex: raise_for_request_error( ex, context=f"OData query {url}", error_cls=ODataError diff --git a/bcb/sgs/__init__.py b/bcb/sgs/__init__.py index 44d81f6..9e5cc84 100644 --- a/bcb/sgs/__init__.py +++ b/bcb/sgs/__init__.py @@ -22,10 +22,12 @@ import pandas as pd from bcb.http import ( + RequestTimeout, get_async_client, get_client, raise_for_request_error, raise_for_status, + timeout_kwargs, with_retry, ) from bcb.exceptions import SGSError @@ -231,13 +233,17 @@ def _raise_sgs_response_error(res: httpx.Response, code: int) -> None: @with_retry -def _get_sgs_response(url: str, payload: Dict[str, str]) -> httpx.Response: - return get_client().get(url, params=payload) +def _get_sgs_response( + url: str, payload: Dict[str, str], timeout: RequestTimeout +) -> httpx.Response: + return get_client().get(url, params=payload, **timeout_kwargs(timeout)) @with_retry -async def _async_get_sgs_response(url: str, payload: Dict[str, str]) -> httpx.Response: - return await get_async_client().get(url, params=payload) +async def _async_get_sgs_response( + url: str, payload: Dict[str, str], timeout: RequestTimeout +) -> httpx.Response: + return await get_async_client().get(url, params=payload, **timeout_kwargs(timeout)) def _format_df(df: pd.DataFrame, code: SGSCode, freq: Optional[str]) -> pd.DataFrame: @@ -262,6 +268,8 @@ def get( multi: bool = ..., freq: Optional[str] = ..., output: Literal["dataframe"] = ..., + *, + timeout: RequestTimeout = ..., ) -> Union[pd.DataFrame, List[pd.DataFrame]]: ... @@ -274,6 +282,8 @@ def get( multi: bool = ..., freq: Optional[str] = ..., output: Literal["text"] = ..., + *, + timeout: RequestTimeout = ..., ) -> Union[str, Dict[int, str]]: ... @@ -285,6 +295,8 @@ def get( multi: bool = True, freq: Optional[str] = None, output: Literal["dataframe", "text"] = "dataframe", + *, + timeout: RequestTimeout = None, ) -> Union[pd.DataFrame, List[pd.DataFrame], str, Dict[int, str]]: """ Retorna um DataFrame pandas com séries temporais obtidas do SGS. @@ -322,6 +334,9 @@ def get( um DataFrame pandas, ou ``'text'`` para retornar o JSON bruto da API do BCB. Para um único código retorna uma string; para múltiplos códigos retorna um ``dict`` mapeando código inteiro → JSON string. + timeout : float or httpx.Timeout, optional + Timeout por tentativa HTTP, em segundos ou como ``httpx.Timeout``. + Quando omitido, usa o timeout padrão do cliente compartilhado. Returns ------- @@ -347,7 +362,9 @@ def get( if output == "text": results: Dict[int, str] = {} for code in code_list: - results[code.value] = get_json(code.value, start, end, last) + results[code.value] = get_json( + code.value, start, end, last, timeout=timeout + ) values = list(results.values()) if len(values) == 1: return values[0] @@ -355,7 +372,7 @@ def get( dfs = [] for code in code_list: - text = get_json(code.value, start, end, last) + text = get_json(code.value, start, end, last, timeout=timeout) df = pd.read_json(StringIO(text)) df = _format_df(df, code, freq) dfs.append(df) @@ -373,6 +390,8 @@ def get_json( start: Optional[DateInput] = None, end: Optional[DateInput] = None, last: int = 0, + *, + timeout: RequestTimeout = None, ) -> str: """ Retorna um JSON com séries temporais obtidas do SGS. @@ -392,6 +411,9 @@ def get_json( Retorna os últimos ``last`` elementos disponíveis da série temporal solicitada. Se ``last`` for maior que 0 (zero) os argumentos ``start`` e ``end`` são ignorados. + timeout : float or httpx.Timeout, optional + Timeout por tentativa HTTP, em segundos ou como ``httpx.Timeout``. + Quando omitido, usa o timeout padrão do cliente compartilhado. Returns ------- @@ -406,7 +428,7 @@ def get_json( f"Fetching SGS time series code={code_obj.value} from {url.split('/dados')[0]}" ) try: - res = _get_sgs_response(url, payload) + res = _get_sgs_response(url, payload, timeout) except httpx.HTTPError as ex: raise_for_request_error( ex, context=f"SGS time series code={code_obj.value}", error_cls=SGSError @@ -423,6 +445,8 @@ async def async_get_json( start: Optional[DateInput] = None, end: Optional[DateInput] = None, last: int = 0, + *, + timeout: RequestTimeout = None, ) -> str: """ Retorna um JSON com séries temporais obtidas do SGS (async version). @@ -439,6 +463,9 @@ async def async_get_json( ``'today'`` e ``'now'`` também são aceitos. last : int Retorna os últimos ``last`` elementos disponíveis + timeout : float or httpx.Timeout, optional + Timeout por tentativa HTTP, em segundos ou como ``httpx.Timeout``. + Quando omitido, usa o timeout padrão do cliente compartilhado. Returns ------- @@ -460,7 +487,7 @@ async def async_get_json( f"from {url.split('/dados')[0]}" ) try: - res = await _async_get_sgs_response(url, payload) + res = await _async_get_sgs_response(url, payload, timeout) except httpx.HTTPError as ex: raise_for_request_error( ex, context=f"SGS time series code={code_obj.value}", error_cls=SGSError @@ -482,6 +509,8 @@ async def async_get( multi: bool = True, freq: Optional[str] = None, output: Literal["dataframe", "text"] = "dataframe", + *, + timeout: RequestTimeout = None, ) -> Union[pd.DataFrame, List[pd.DataFrame], str, Dict[int, str]]: """ Retorna um DataFrame pandas com séries temporais obtidas do SGS (async version). @@ -507,6 +536,9 @@ async def async_get( Frequência a ser utilizada na série temporal output : str Formato de saída: ``'dataframe'`` ou ``'text'`` + timeout : float or httpx.Timeout, optional + Timeout por tentativa HTTP, em segundos ou como ``httpx.Timeout``. + Quando omitido, usa o timeout padrão do cliente compartilhado. Returns ------- @@ -518,7 +550,7 @@ async def async_get( # Concurrent HTTP requests via asyncio.gather() texts = await asyncio.gather( - *[async_get_json(c.value, start, end, last) for c in code_list] + *[async_get_json(c.value, start, end, last, timeout=timeout) for c in code_list] ) if output == "text": diff --git a/docs/currency.rst b/docs/currency.rst index eb75822..682fa72 100644 --- a/docs/currency.rst +++ b/docs/currency.rst @@ -60,6 +60,18 @@ Conversor de Moedas O módulo :py:mod:`bcb.currency` obtém dados de moedas do conversor de moedas do Banco Central através de webscraping. Os parâmetros ``start`` e ``end`` aceitam strings ``YYYY-MM-DD``, ``datetime.date``, ``datetime.datetime`` ou :py:class:`bcb.utils.Date`. +Para consultas longas ou respostas lentas, informe ``timeout`` na chamada. O valor +é usado em cada requisição HTTP feita por ``currency.get`` na chamada corrente, +incluindo a busca inicial das listas de moedas quando elas ainda não estiverem +em cache. + +.. code:: python + + from bcb import currency + + df = currency.get("USD", start="1980-01-01", end="2026-01-01", timeout=120) + moedas = currency.get_currency_list(timeout=120) + .. ipython:: python from bcb import currency diff --git a/docs/odata.rst b/docs/odata.rst index aac1c15..ffde87c 100644 --- a/docs/odata.rst +++ b/docs/odata.rst @@ -21,6 +21,29 @@ a API a partir da URL da API que está disponível no `portal de dados abertos < do Banco Central. Veja mais detalhes em :ref:`Classe ODataAPI`. +Timeout por chamada +------------------- + +As APIs OData também aceitam ``timeout`` para serviços lentos ou consultas com +muitos registros. O timeout informado no construtor é usado na descoberta do +serviço, no carregamento dos metadados e como padrão das queries criadas a +partir dessa API. Métodos de execução podem receber outro valor para sobrescrever +o padrão apenas naquela chamada. + +.. code:: python + + from bcb import Expectativas + + em = Expectativas(timeout=120) + ep = em.get_endpoint("ExpectativasMercadoAnuais") + + df = ep.get(timeout=60) + raw = ep.query().limit(1000).text(timeout=60) + data = ep.query().limit(1000).collect(timeout=60) + +O mesmo parâmetro está disponível nas versões assíncronas ``async_get``, +``async_text`` e ``async_collect``. + Segue um exemplo de como acessar a API do PIX. .. ipython:: python diff --git a/docs/sgs.rst b/docs/sgs.rst index 5e31118..f4f3496 100644 --- a/docs/sgs.rst +++ b/docs/sgs.rst @@ -7,6 +7,25 @@ interface JSON do serviço BCData/SGS - Os parâmetros ``start`` e ``end`` aceitam strings ``YYYY-MM-DD``, ``datetime.date``, ``datetime.datetime`` ou :py:class:`bcb.utils.Date`. Também é possível usar ``last`` para buscar os últimos ``n`` pontos disponíveis. +Timeout em consultas longas +--------------------------- + +Por padrão, as requisições usam o timeout global do cliente HTTP compartilhado. +Para consultas SGS com janelas grandes ou respostas lentas, informe ``timeout`` +na chamada. O valor é aplicado por tentativa HTTP; quando houver retry, cada +tentativa usa o mesmo timeout. + +.. code:: python + + from bcb import sgs + + df = sgs.get(11, start="1990-01-01", end="2026-01-01", timeout=120) + raw = sgs.get_json(11, start="1990-01-01", timeout=120) + +Se a consulta continuar lenta mesmo com timeout maior, divida o período em +janelas menores e concatene os resultados. + + Exemplos -------- diff --git a/tests/test_request_timeout.py b/tests/test_request_timeout.py new file mode 100644 index 0000000..f3d4381 --- /dev/null +++ b/tests/test_request_timeout.py @@ -0,0 +1,192 @@ +from datetime import datetime +from typing import Any + +import httpx +import pandas as pd +import pytest + +from bcb import currency, sgs +from bcb import http as http_module +from bcb.odata import framework as odata_framework +from bcb.odata.api import Expectativas +from tests.conftest import ( + CURRENCY_ID_LIST_HTML, + CURRENCY_LIST_CSV, + CURRENCY_RATE_CSV, + ODATA_METADATA_XML, + ODATA_QUERY_RESPONSE_JSON, + ODATA_SERVICE_ROOT_JSON, + SGS_JSON_5, +) + +START = datetime(2020, 12, 1) +END = datetime(2020, 12, 7) +EXPECTATIVAS_BASE_URL = ( + "https://olinda.bcb.gov.br/olinda/servico/Expectativas/versao/v1/odata/" +) + + +class RecordingClient: + def __init__(self) -> None: + self.calls: list[tuple[str, dict[str, Any]]] = [] + + def get(self, url: str, **kwargs: Any) -> httpx.Response: + self.calls.append((url, kwargs.copy())) + return response_for_url(url) + + +class AsyncRecordingClient(RecordingClient): + async def get(self, url: str, **kwargs: Any) -> httpx.Response: + self.calls.append((url, kwargs.copy())) + return response_for_url(url) + + +def response_for_url(url: str) -> httpx.Response: + request = httpx.Request("GET", url) + if "bcdata.sgs" in url: + return httpx.Response(200, text=SGS_JSON_5, request=request) + if "exibeFormularioConsultaBoletim" in url: + return httpx.Response(200, content=CURRENCY_ID_LIST_HTML, request=request) + if "Download/fechamento" in url: + return httpx.Response(200, text=CURRENCY_LIST_CSV, request=request) + if "gerarCSVFechamento" in url: + return httpx.Response( + 200, + text=CURRENCY_RATE_CSV, + headers={"Content-Type": "text/csv"}, + request=request, + ) + if url == EXPECTATIVAS_BASE_URL: + return httpx.Response(200, text=ODATA_SERVICE_ROOT_JSON, request=request) + if url.endswith("$" + "metadata"): + return httpx.Response(200, content=ODATA_METADATA_XML, request=request) + if "ExpectativasMercadoAnuais" in url: + return httpx.Response(200, text=ODATA_QUERY_RESPONSE_JSON, request=request) + raise AssertionError(f"unexpected URL: {url}") + + +def timeout_values(client: RecordingClient) -> list[Any]: + return [kwargs.get("timeout") for _, kwargs in client.calls] + + +def test_timeout_kwargs_omits_none_to_keep_client_default() -> None: + assert http_module.timeout_kwargs(None) == {} + assert http_module.timeout_kwargs(45.0) == {"timeout": 45.0} + + +def test_sgs_get_passes_timeout_to_request(monkeypatch: pytest.MonkeyPatch) -> None: + client = RecordingClient() + monkeypatch.setattr(sgs, "get_client", lambda: client) + + df = sgs.get(1, timeout=45.0) + + assert isinstance(df, pd.DataFrame) + assert timeout_values(client) == [45.0] + + +def test_sgs_get_uses_client_default_when_timeout_omitted( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = RecordingClient() + monkeypatch.setattr(sgs, "get_client", lambda: client) + + sgs.get_json(1) + + assert "timeout" not in client.calls[0][1] + + +def test_currency_get_passes_timeout_to_all_initial_requests( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = RecordingClient() + monkeypatch.setattr(currency, "get_client", lambda: client) + + df = currency.get("USD", START, END, timeout=60.0) + + assert isinstance(df, pd.DataFrame) + assert len(client.calls) == 3 + assert timeout_values(client) == [60.0, 60.0, 60.0] + + +def test_currency_get_currency_list_passes_timeout( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = RecordingClient() + monkeypatch.setattr(currency, "get_client", lambda: client) + + df = currency.get_currency_list(timeout=15.0) + + assert isinstance(df, pd.DataFrame) + assert timeout_values(client) == [15.0] + + +def test_odata_api_and_endpoint_pass_timeouts( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = RecordingClient() + monkeypatch.setattr(odata_framework, "get_client", lambda: client) + + api = Expectativas(timeout=12.0) + endpoint = api.get_endpoint("ExpectativasMercadoAnuais") + data = endpoint.get(timeout=24.0) + + assert isinstance(data, pd.DataFrame) + assert timeout_values(client) == [12.0, 12.0, 24.0] + + +def test_odata_query_uses_api_timeout_by_default( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = RecordingClient() + monkeypatch.setattr(odata_framework, "get_client", lambda: client) + + api = Expectativas(timeout=18.0) + endpoint = api.get_endpoint("ExpectativasMercadoAnuais") + endpoint.query().limit(1).text() + + assert timeout_values(client) == [18.0, 18.0, 18.0] + + +@pytest.mark.anyio +async def test_async_sgs_get_json_passes_timeout( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = AsyncRecordingClient() + monkeypatch.setattr(sgs, "get_async_client", lambda: client) + + result = await sgs.async_get_json(1, timeout=30.0) + + assert result == SGS_JSON_5 + assert timeout_values(client) == [30.0] + + +@pytest.mark.anyio +async def test_async_currency_get_passes_timeout_to_all_initial_requests( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = AsyncRecordingClient() + monkeypatch.setattr(currency, "get_async_client", lambda: client) + + df = await currency.async_get("USD", START, END, timeout=35.0) + + assert isinstance(df, pd.DataFrame) + assert len(client.calls) == 3 + assert timeout_values(client) == [35.0, 35.0, 35.0] + + +@pytest.mark.anyio +async def test_async_odata_endpoint_passes_timeout( + monkeypatch: pytest.MonkeyPatch, +) -> None: + sync_client = RecordingClient() + async_client = AsyncRecordingClient() + monkeypatch.setattr(odata_framework, "get_client", lambda: sync_client) + monkeypatch.setattr(odata_framework, "get_async_client", lambda: async_client) + + api = Expectativas(timeout=14.0) + endpoint = api.get_endpoint("ExpectativasMercadoAnuais") + data = await endpoint.async_get(timeout=28.0) + + assert isinstance(data, pd.DataFrame) + assert timeout_values(sync_client) == [14.0, 14.0] + assert timeout_values(async_client) == [28.0]