Skip to content

API Clients

[!NOTE] For most users, the access layer is recommended. This section documents the low-level API client interface that returns raw dictionaries. The access layer (documented in access layer) returns pandas DataFrames and is easier to use for data analysis.

Use the API layer (bdl.api.*) when you need: - Raw API response structure - Custom response processing - Direct access to API metadata - Integration with non-pandas workflows

Use the access layer (bdl.*) when you need: - Pandas DataFrames for data analysis - Automatic column normalization and type inference - Nested data flattening

The pyBDL library provides a comprehensive set of API clients for interacting with the Local Data Bank (BDL) API. All API endpoints are accessible through the main client's .api attribute. See Main client for details about the main client.

Endpoint Class Description
Aggregates pybdl.api.aggregates.AggregatesAPI Aggregation level metadata and details
Attributes pybdl.api.attributes.AttributesAPI Attribute metadata and details
Data pybdl.api.data.DataAPI Statistical data access (variables, units, localities)
Levels pybdl.api.levels.LevelsAPI Administrative unit aggregation levels
Measures pybdl.api.measures.MeasuresAPI Measure unit metadata
Subjects pybdl.api.subjects.SubjectsAPI Subject hierarchy and metadata
Units pybdl.api.units.UnitsAPI Administrative unit metadata
Variables pybdl.api.variables.VariablesAPI Variable metadata and details
Version pybdl.api.version.VersionAPI API version and build info
Years pybdl.api.years.YearsAPI Available years for data

Available API Endpoints

[!NOTE] All API clients are accessible via bdl.api.<endpoint> (e.g., bdl.api.data.get_data_by_variable(...)).

Seealso

Async Usage

All API clients support async methods for high-performance and concurrent applications. Async methods are named with an a prefix (e.g., aget_data_by_variable).

import asyncio
from pybdl import BDL

async def main():
    bdl = BDL()
    data = await bdl.api.data.aget_data_by_variable(variable_id="3643", years=[2021])
    print(data)

asyncio.run(main())

[!NOTE] Async methods are available for all endpoints. See the API reference below for details.

Format and Language Parameters

API clients support format and language parameters for controlling response content:

Format Options (FormatLiteral): - "json" - JSON format (default) - "jsonapi" - JSON:API format - "xml" - XML format

Language Options (LanguageLiteral): - "pl" - Polish (default if configured) - "en" - English

The format and language parameters automatically set the appropriate HTTP headers: - Accept header is set based on the format parameter - Accept-Language header is set based on the language parameter

from pybdl import BDL

bdl = BDL()

# Request data in XML format
data = bdl.api.data.get_data_by_variable(
    variable_id="3643",
    years=[2021],
    format="xml"
)

# Request data in Polish
data = bdl.api.data.get_data_by_variable(
    variable_id="3643",
    years=[2021],
    lang="pl"
)

Aggregates

aggregates

AggregatesAPI

AggregatesAPI(config, extra_headers=None)

Bases: BaseAPIClient

Client for the BDL /aggregates endpoints.

Initialize base API client for BDL.

Parameters:

Name Type Description Default
config BDLConfig

BDL configuration object.

required
extra_headers dict[str, str] | None

Optional extra headers (e.g., Accept-Language) to include in requests.

None
Source code in pybdl/api/client.py
def __init__(self, config: BDLConfig, extra_headers: dict[str, str] | None = None):
    """
    Initialize base API client for BDL.

    Args:
        config: BDL configuration object.
        extra_headers: Optional extra headers (e.g., Accept-Language) to include in requests.
    """
    self.config = config
    is_registered = bool(config.api_key)
    quotas: QuotaMap = cast(
        QuotaMap,
        config.custom_quotas if config.custom_quotas is not None else DEFAULT_QUOTAS,
    )
    self._quota_cache = PersistentQuotaCache(
        config.quota_cache_enabled,
        cache_file=config.quota_cache_file,
        use_global_cache=config.use_global_cache,
    )
    self._sync_limiter = RateLimiter(
        quotas,
        is_registered,
        self._quota_cache,
        raise_on_limit=config.raise_on_rate_limit,
    )
    self._async_limiter = AsyncRateLimiter(
        quotas,
        is_registered,
        self._quota_cache,
        raise_on_limit=config.raise_on_rate_limit,
    )
    self._proxy_url = self._build_proxy_url()
    self._http_cache_path = resolve_http_cache_db_path(config.cache_backend, self._quota_cache.cache_file)
    default_headers = self._build_default_headers(extra_headers)
    self.session = build_sync_http_client(
        cache_backend=config.cache_backend,
        http_cache_db_path=self._http_cache_path,
        default_headers=default_headers,
        proxy=self._proxy_url,
    )
    self._async_client = build_async_http_client(
        cache_backend=config.cache_backend,
        http_cache_db_path=self._http_cache_path,
        default_headers=default_headers,
        proxy=self._proxy_url,
    )

config instance-attribute

config = config

session instance-attribute

session = build_sync_http_client(
    cache_backend=cache_backend,
    http_cache_db_path=_http_cache_path,
    default_headers=default_headers,
    proxy=_proxy_url,
)

list_aggregates

list_aggregates(
    sort=None,
    page_size=100,
    max_pages=None,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/aggregates.py
def list_aggregates(
    self,
    sort: str | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    return self._fetch_collection_endpoint(
        "aggregates",
        extra_params=self._list_params(sort, extra_query),
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
        page_size=page_size,
        max_pages=max_pages,
    )

get_aggregate

get_aggregate(
    aggregate_id,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/aggregates.py
def get_aggregate(
    self,
    aggregate_id: str,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return self._fetch_detail_endpoint(
        f"aggregates/{aggregate_id}",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

get_aggregates_metadata

get_aggregates_metadata(
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/aggregates.py
def get_aggregates_metadata(
    self,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return self._fetch_detail_endpoint(
        "aggregates/metadata",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

alist_aggregates async

alist_aggregates(
    sort=None,
    page_size=100,
    max_pages=None,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/aggregates.py
async def alist_aggregates(
    self,
    sort: str | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    return await self._afetch_collection_endpoint(
        "aggregates",
        extra_params=self._list_params(sort, extra_query),
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
        page_size=page_size,
        max_pages=max_pages,
    )

aget_aggregate async

aget_aggregate(
    aggregate_id,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/aggregates.py
async def aget_aggregate(
    self,
    aggregate_id: str,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return await self._afetch_detail_endpoint(
        f"aggregates/{aggregate_id}",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

aget_aggregates_metadata async

aget_aggregates_metadata(
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/aggregates.py
async def aget_aggregates_metadata(
    self,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return await self._afetch_detail_endpoint(
        "aggregates/metadata",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

close

close()

Close synchronous HTTP resources.

Source code in pybdl/api/client.py
def close(self) -> None:
    """Close synchronous HTTP resources."""
    self.session.close()

aclose async

aclose()

Close synchronous and asynchronous HTTP resources.

Source code in pybdl/api/client.py
async def aclose(self) -> None:
    """Close synchronous and asynchronous HTTP resources."""
    self.close()
    await self._async_client.aclose()

__enter__

__enter__()
Source code in pybdl/api/client.py
def __enter__(self) -> "BaseAPIClient":
    return self

__exit__

__exit__(exc_type, exc_val, exc_tb)
Source code in pybdl/api/client.py
def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> Literal[False]:
    self.close()
    return False

__aenter__ async

__aenter__()
Source code in pybdl/api/client.py
async def __aenter__(self) -> "BaseAPIClient":
    return self

__aexit__ async

__aexit__(exc_type, exc_val, exc_tb)
Source code in pybdl/api/client.py
async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> Literal[False]:
    await self.aclose()
    return False

fetch_all_results

fetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[False] = False,
    show_progress: bool = True,
) -> list[dict[str, Any]]
fetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[True],
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
fetch_all_results(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    return_metadata=False,
    show_progress=True,
)

Fetch paginated results synchronously and combine them into a single list.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
method str

HTTP method (default: GET).

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
results_key str

Key for extracting data from each page.

'results'
page_size int

Items per page.

100
max_pages int | None

Optional limit of pages.

None
return_metadata bool

If True, return (results, metadata).

False
show_progress bool

Display progress via tqdm.

True

Returns:

Type Description
list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Combined list of results, optionally with metadata.

Source code in pybdl/api/client.py
def fetch_all_results(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: bool = False,
    show_progress: bool = True,
) -> list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    """
    Fetch paginated results synchronously and combine them into a single list.

    Args:
        endpoint: API endpoint.
        method: HTTP method (default: GET).
        params: Query parameters.
        headers: Optional request headers.
        results_key: Key for extracting data from each page.
        page_size: Items per page.
        max_pages: Optional limit of pages.
        return_metadata: If True, return (results, metadata).
        show_progress: Display progress via tqdm.

    Returns:
        Combined list of results, optionally with metadata.
    """
    all_results: list[dict[str, Any]] = []
    metadata: dict[str, Any] = {}
    progress_bar = (
        tqdm(desc=f"Fetching {endpoint.split('/')[-1]}", unit=" pages", leave=True) if show_progress else None
    )

    first_page = True
    try:
        for page in self._paginated_request_sync(
            endpoint,
            method=method,
            params=params,
            headers=headers,
            results_key=results_key,
            page_size=page_size,
            max_pages=max_pages,
        ):
            if results_key not in page:
                raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=page)
            if first_page and return_metadata:
                metadata = self._metadata_from_response(page, results_key)
                if progress_bar is not None and "totalCount" in page:
                    total_pages = (page["totalCount"] + page_size - 1) // page_size
                    total_pages = min(total_pages, max_pages) if max_pages else total_pages
                    progress_bar.total = total_pages
                first_page = False

            all_results.extend(page.get(results_key, []))

            if progress_bar is not None:
                progress_bar.update(1)
                progress_bar.set_postfix({"items": len(all_results)})
    finally:
        if progress_bar is not None:
            progress_bar.close()

    return (all_results, metadata) if return_metadata else all_results

fetch_all_results_with_metadata

fetch_all_results_with_metadata(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    show_progress=True,
)
Source code in pybdl/api/client.py
def fetch_all_results_with_metadata(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    return self.fetch_all_results(
        endpoint,
        method=method,
        params=params,
        headers=headers,
        results_key=results_key,
        page_size=page_size,
        max_pages=max_pages,
        return_metadata=True,
        show_progress=show_progress,
    )

fetch_single_result

fetch_single_result(
    endpoint: str,
    *,
    results_key: None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> dict[str, Any]
fetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> list[dict[str, Any]]
fetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[True],
) -> tuple[list[dict[str, Any]], dict[str, Any]]
fetch_single_result(
    endpoint: str,
    *,
    results_key: None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[True],
) -> tuple[dict[str, Any], dict[str, Any]]
fetch_single_result(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
    return_metadata=False,
)

Fetch a single result, non-paginated (sync).

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
results_key str | None

If not None, extract this key from the JSON.

None
method str

HTTP method.

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
return_metadata bool

Also return metadata if True.

False

Returns:

Type Description
dict[str, Any] | list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]] | tuple[dict[str, Any], dict[str, Any]]

Dictionary or list, optionally with separate metadata.

Source code in pybdl/api/client.py
def fetch_single_result(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: bool = False,
) -> (
    dict[str, Any]
    | list[dict[str, Any]]
    | tuple[list[dict[str, Any]], dict[str, Any]]
    | tuple[dict[str, Any], dict[str, Any]]
):
    """
    Fetch a single result, non-paginated (sync).

    Args:
        endpoint: API endpoint.
        results_key: If not None, extract this key from the JSON.
        method: HTTP method.
        params: Query parameters.
        headers: Optional request headers.
        return_metadata: Also return metadata if True.

    Returns:
        Dictionary or list, optionally with separate metadata.
    """
    response = self._request_sync(
        endpoint=endpoint,
        method=method,
        params=params,
        headers=headers,
    )

    if results_key is None:
        if return_metadata:
            return response, {}
        return response

    if not isinstance(response, dict) or results_key not in response:
        raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=response)

    results_val = response[results_key]
    if return_metadata:
        metadata = self._metadata_from_response(response, results_key)
        return results_val, metadata

    return results_val

fetch_single_result_with_metadata

fetch_single_result_with_metadata(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
)
Source code in pybdl/api/client.py
def fetch_single_result_with_metadata(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    return self.fetch_single_result(
        endpoint,
        results_key=results_key,
        method=method,
        params=params,
        headers=headers,
        return_metadata=True,
    )

afetch_all_results async

afetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[False] = False,
    show_progress: bool = True,
) -> list[dict[str, Any]]
afetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[True],
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
afetch_all_results(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    return_metadata=False,
    show_progress=True,
)

Asynchronously fetch paginated results and combine them into a single list.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
method str

HTTP method (default: GET).

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
results_key str

Key for extracting data from each page.

'results'
page_size int

Items per page.

100
max_pages int | None

Optional limit of pages.

None
return_metadata bool

If True, return (results, metadata).

False
show_progress bool

Display progress via tqdm.

True

Returns:

Type Description
list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Combined list of results, optionally with metadata.

Source code in pybdl/api/client.py
async def afetch_all_results(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: bool = False,
    show_progress: bool = True,
) -> list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    """
    Asynchronously fetch paginated results and combine them into a single list.

    Args:
        endpoint: API endpoint.
        method: HTTP method (default: GET).
        params: Query parameters.
        headers: Optional request headers.
        results_key: Key for extracting data from each page.
        page_size: Items per page.
        max_pages: Optional limit of pages.
        return_metadata: If True, return (results, metadata).
        show_progress: Display progress via tqdm.

    Returns:
        Combined list of results, optionally with metadata.
    """
    all_results: list[dict[str, Any]] = []
    metadata: dict[str, Any] = {}
    first_page = True
    progress_bar = (
        tqdm(desc=f"Fetching {endpoint.split('/')[-1]} (async)", unit=" pages", leave=True)
        if show_progress
        else None
    )

    try:
        async for page in self._paginated_request_async(
            endpoint,
            method=method,
            params=params,
            headers=headers,
            results_key=results_key,
            page_size=page_size,
            max_pages=max_pages,
        ):
            if results_key not in page:
                raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=page)
            if first_page and return_metadata:
                metadata = self._metadata_from_response(page, results_key)
                if progress_bar is not None and "totalCount" in page:
                    total_pages = (page["totalCount"] + page_size - 1) // page_size
                    total_pages = min(total_pages, max_pages) if max_pages else total_pages
                    progress_bar.total = total_pages
                first_page = False

            all_results.extend(page.get(results_key, []))
            if progress_bar is not None:
                progress_bar.update(1)
                progress_bar.set_postfix({"items": len(all_results)})
    finally:
        if progress_bar is not None:
            progress_bar.close()

    return (all_results, metadata) if return_metadata else all_results

afetch_all_results_with_metadata async

afetch_all_results_with_metadata(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    show_progress=True,
)
Source code in pybdl/api/client.py
async def afetch_all_results_with_metadata(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    return await self.afetch_all_results(
        endpoint,
        method=method,
        params=params,
        headers=headers,
        results_key=results_key,
        page_size=page_size,
        max_pages=max_pages,
        return_metadata=True,
        show_progress=show_progress,
    )

afetch_single_result async

afetch_single_result(
    endpoint: str,
    *,
    method: str = "GET",
    results_key: Literal[None] = None,
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> dict[str, Any]
afetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> list[dict[str, Any]]
afetch_single_result(
    endpoint: str,
    *,
    return_metadata: Literal[True],
    results_key: Literal[None] = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]]
afetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    return_metadata: Literal[True],
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
afetch_single_result(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
    return_metadata=False,
)

Asynchronously fetch a single result, non-paginated.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
results_key str | None

If not None, extract this key from the JSON.

None
method str

HTTP method.

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
return_metadata bool

Also return metadata if True.

False

Returns:

Type Description
dict[str, Any] | list[dict[str, Any]] | tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Dictionary or list, optionally with separate metadata.

Source code in pybdl/api/client.py
async def afetch_single_result(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: bool = False,
) -> (
    dict[str, Any]
    | list[dict[str, Any]]
    | tuple[dict[str, Any], dict[str, Any]]
    | tuple[list[dict[str, Any]], dict[str, Any]]
):
    """
    Asynchronously fetch a single result, non-paginated.

    Args:
        endpoint: API endpoint.
        results_key: If not None, extract this key from the JSON.
        method: HTTP method.
        params: Query parameters.
        headers: Optional request headers.
        return_metadata: Also return metadata if True.

    Returns:
        Dictionary or list, optionally with separate metadata.
    """
    response = await self._request_async(
        endpoint=endpoint,
        method=method,
        params=params,
        headers=headers,
    )

    if results_key is None:
        return (response, {}) if return_metadata else response

    if not isinstance(response, dict) or results_key not in response:
        raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=response)

    results_val = cast(list[dict[str, Any]], response[results_key])
    if return_metadata:
        metadata = self._metadata_from_response(response, results_key)
        return results_val, metadata

    return results_val

afetch_single_result_with_metadata async

afetch_single_result_with_metadata(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
)
Source code in pybdl/api/client.py
async def afetch_single_result_with_metadata(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    return await self.afetch_single_result(
        endpoint,
        results_key=results_key,
        method=method,
        params=params,
        headers=headers,
        return_metadata=True,
    )

Attributes

attributes

AttributesAPI

AttributesAPI(config, extra_headers=None)

Bases: BaseAPIClient

Client for the BDL /attributes endpoints.

Initialize base API client for BDL.

Parameters:

Name Type Description Default
config BDLConfig

BDL configuration object.

required
extra_headers dict[str, str] | None

Optional extra headers (e.g., Accept-Language) to include in requests.

None
Source code in pybdl/api/client.py
def __init__(self, config: BDLConfig, extra_headers: dict[str, str] | None = None):
    """
    Initialize base API client for BDL.

    Args:
        config: BDL configuration object.
        extra_headers: Optional extra headers (e.g., Accept-Language) to include in requests.
    """
    self.config = config
    is_registered = bool(config.api_key)
    quotas: QuotaMap = cast(
        QuotaMap,
        config.custom_quotas if config.custom_quotas is not None else DEFAULT_QUOTAS,
    )
    self._quota_cache = PersistentQuotaCache(
        config.quota_cache_enabled,
        cache_file=config.quota_cache_file,
        use_global_cache=config.use_global_cache,
    )
    self._sync_limiter = RateLimiter(
        quotas,
        is_registered,
        self._quota_cache,
        raise_on_limit=config.raise_on_rate_limit,
    )
    self._async_limiter = AsyncRateLimiter(
        quotas,
        is_registered,
        self._quota_cache,
        raise_on_limit=config.raise_on_rate_limit,
    )
    self._proxy_url = self._build_proxy_url()
    self._http_cache_path = resolve_http_cache_db_path(config.cache_backend, self._quota_cache.cache_file)
    default_headers = self._build_default_headers(extra_headers)
    self.session = build_sync_http_client(
        cache_backend=config.cache_backend,
        http_cache_db_path=self._http_cache_path,
        default_headers=default_headers,
        proxy=self._proxy_url,
    )
    self._async_client = build_async_http_client(
        cache_backend=config.cache_backend,
        http_cache_db_path=self._http_cache_path,
        default_headers=default_headers,
        proxy=self._proxy_url,
    )

config instance-attribute

config = config

session instance-attribute

session = build_sync_http_client(
    cache_backend=cache_backend,
    http_cache_db_path=_http_cache_path,
    default_headers=default_headers,
    proxy=_proxy_url,
)

list_attributes

list_attributes(
    sort=None,
    page_size=100,
    max_pages=None,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/attributes.py
def list_attributes(
    self,
    sort: str | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    return self._fetch_collection_endpoint(
        "attributes",
        extra_params=self._list_params(sort, extra_query),
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
        page_size=page_size,
        max_pages=max_pages,
    )

get_attribute

get_attribute(
    attribute_id,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/attributes.py
def get_attribute(
    self,
    attribute_id: str,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return self._fetch_detail_endpoint(
        f"attributes/{attribute_id}",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

get_attributes_metadata

get_attributes_metadata(
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/attributes.py
def get_attributes_metadata(
    self,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return self._fetch_detail_endpoint(
        "attributes/metadata",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

alist_attributes async

alist_attributes(
    sort=None,
    page_size=100,
    max_pages=None,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/attributes.py
async def alist_attributes(
    self,
    sort: str | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    return await self._afetch_collection_endpoint(
        "attributes",
        extra_params=self._list_params(sort, extra_query),
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
        page_size=page_size,
        max_pages=max_pages,
    )

aget_attribute async

aget_attribute(
    attribute_id,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/attributes.py
async def aget_attribute(
    self,
    attribute_id: str,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return await self._afetch_detail_endpoint(
        f"attributes/{attribute_id}",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

aget_attributes_metadata async

aget_attributes_metadata(
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/attributes.py
async def aget_attributes_metadata(
    self,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return await self._afetch_detail_endpoint(
        "attributes/metadata",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

close

close()

Close synchronous HTTP resources.

Source code in pybdl/api/client.py
def close(self) -> None:
    """Close synchronous HTTP resources."""
    self.session.close()

aclose async

aclose()

Close synchronous and asynchronous HTTP resources.

Source code in pybdl/api/client.py
async def aclose(self) -> None:
    """Close synchronous and asynchronous HTTP resources."""
    self.close()
    await self._async_client.aclose()

__enter__

__enter__()
Source code in pybdl/api/client.py
def __enter__(self) -> "BaseAPIClient":
    return self

__exit__

__exit__(exc_type, exc_val, exc_tb)
Source code in pybdl/api/client.py
def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> Literal[False]:
    self.close()
    return False

__aenter__ async

__aenter__()
Source code in pybdl/api/client.py
async def __aenter__(self) -> "BaseAPIClient":
    return self

__aexit__ async

__aexit__(exc_type, exc_val, exc_tb)
Source code in pybdl/api/client.py
async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> Literal[False]:
    await self.aclose()
    return False

fetch_all_results

fetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[False] = False,
    show_progress: bool = True,
) -> list[dict[str, Any]]
fetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[True],
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
fetch_all_results(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    return_metadata=False,
    show_progress=True,
)

Fetch paginated results synchronously and combine them into a single list.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
method str

HTTP method (default: GET).

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
results_key str

Key for extracting data from each page.

'results'
page_size int

Items per page.

100
max_pages int | None

Optional limit of pages.

None
return_metadata bool

If True, return (results, metadata).

False
show_progress bool

Display progress via tqdm.

True

Returns:

Type Description
list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Combined list of results, optionally with metadata.

Source code in pybdl/api/client.py
def fetch_all_results(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: bool = False,
    show_progress: bool = True,
) -> list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    """
    Fetch paginated results synchronously and combine them into a single list.

    Args:
        endpoint: API endpoint.
        method: HTTP method (default: GET).
        params: Query parameters.
        headers: Optional request headers.
        results_key: Key for extracting data from each page.
        page_size: Items per page.
        max_pages: Optional limit of pages.
        return_metadata: If True, return (results, metadata).
        show_progress: Display progress via tqdm.

    Returns:
        Combined list of results, optionally with metadata.
    """
    all_results: list[dict[str, Any]] = []
    metadata: dict[str, Any] = {}
    progress_bar = (
        tqdm(desc=f"Fetching {endpoint.split('/')[-1]}", unit=" pages", leave=True) if show_progress else None
    )

    first_page = True
    try:
        for page in self._paginated_request_sync(
            endpoint,
            method=method,
            params=params,
            headers=headers,
            results_key=results_key,
            page_size=page_size,
            max_pages=max_pages,
        ):
            if results_key not in page:
                raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=page)
            if first_page and return_metadata:
                metadata = self._metadata_from_response(page, results_key)
                if progress_bar is not None and "totalCount" in page:
                    total_pages = (page["totalCount"] + page_size - 1) // page_size
                    total_pages = min(total_pages, max_pages) if max_pages else total_pages
                    progress_bar.total = total_pages
                first_page = False

            all_results.extend(page.get(results_key, []))

            if progress_bar is not None:
                progress_bar.update(1)
                progress_bar.set_postfix({"items": len(all_results)})
    finally:
        if progress_bar is not None:
            progress_bar.close()

    return (all_results, metadata) if return_metadata else all_results

fetch_all_results_with_metadata

fetch_all_results_with_metadata(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    show_progress=True,
)
Source code in pybdl/api/client.py
def fetch_all_results_with_metadata(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    return self.fetch_all_results(
        endpoint,
        method=method,
        params=params,
        headers=headers,
        results_key=results_key,
        page_size=page_size,
        max_pages=max_pages,
        return_metadata=True,
        show_progress=show_progress,
    )

fetch_single_result

fetch_single_result(
    endpoint: str,
    *,
    results_key: None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> dict[str, Any]
fetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> list[dict[str, Any]]
fetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[True],
) -> tuple[list[dict[str, Any]], dict[str, Any]]
fetch_single_result(
    endpoint: str,
    *,
    results_key: None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[True],
) -> tuple[dict[str, Any], dict[str, Any]]
fetch_single_result(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
    return_metadata=False,
)

Fetch a single result, non-paginated (sync).

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
results_key str | None

If not None, extract this key from the JSON.

None
method str

HTTP method.

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
return_metadata bool

Also return metadata if True.

False

Returns:

Type Description
dict[str, Any] | list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]] | tuple[dict[str, Any], dict[str, Any]]

Dictionary or list, optionally with separate metadata.

Source code in pybdl/api/client.py
def fetch_single_result(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: bool = False,
) -> (
    dict[str, Any]
    | list[dict[str, Any]]
    | tuple[list[dict[str, Any]], dict[str, Any]]
    | tuple[dict[str, Any], dict[str, Any]]
):
    """
    Fetch a single result, non-paginated (sync).

    Args:
        endpoint: API endpoint.
        results_key: If not None, extract this key from the JSON.
        method: HTTP method.
        params: Query parameters.
        headers: Optional request headers.
        return_metadata: Also return metadata if True.

    Returns:
        Dictionary or list, optionally with separate metadata.
    """
    response = self._request_sync(
        endpoint=endpoint,
        method=method,
        params=params,
        headers=headers,
    )

    if results_key is None:
        if return_metadata:
            return response, {}
        return response

    if not isinstance(response, dict) or results_key not in response:
        raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=response)

    results_val = response[results_key]
    if return_metadata:
        metadata = self._metadata_from_response(response, results_key)
        return results_val, metadata

    return results_val

fetch_single_result_with_metadata

fetch_single_result_with_metadata(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
)
Source code in pybdl/api/client.py
def fetch_single_result_with_metadata(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    return self.fetch_single_result(
        endpoint,
        results_key=results_key,
        method=method,
        params=params,
        headers=headers,
        return_metadata=True,
    )

afetch_all_results async

afetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[False] = False,
    show_progress: bool = True,
) -> list[dict[str, Any]]
afetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[True],
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
afetch_all_results(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    return_metadata=False,
    show_progress=True,
)

Asynchronously fetch paginated results and combine them into a single list.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
method str

HTTP method (default: GET).

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
results_key str

Key for extracting data from each page.

'results'
page_size int

Items per page.

100
max_pages int | None

Optional limit of pages.

None
return_metadata bool

If True, return (results, metadata).

False
show_progress bool

Display progress via tqdm.

True

Returns:

Type Description
list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Combined list of results, optionally with metadata.

Source code in pybdl/api/client.py
async def afetch_all_results(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: bool = False,
    show_progress: bool = True,
) -> list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    """
    Asynchronously fetch paginated results and combine them into a single list.

    Args:
        endpoint: API endpoint.
        method: HTTP method (default: GET).
        params: Query parameters.
        headers: Optional request headers.
        results_key: Key for extracting data from each page.
        page_size: Items per page.
        max_pages: Optional limit of pages.
        return_metadata: If True, return (results, metadata).
        show_progress: Display progress via tqdm.

    Returns:
        Combined list of results, optionally with metadata.
    """
    all_results: list[dict[str, Any]] = []
    metadata: dict[str, Any] = {}
    first_page = True
    progress_bar = (
        tqdm(desc=f"Fetching {endpoint.split('/')[-1]} (async)", unit=" pages", leave=True)
        if show_progress
        else None
    )

    try:
        async for page in self._paginated_request_async(
            endpoint,
            method=method,
            params=params,
            headers=headers,
            results_key=results_key,
            page_size=page_size,
            max_pages=max_pages,
        ):
            if results_key not in page:
                raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=page)
            if first_page and return_metadata:
                metadata = self._metadata_from_response(page, results_key)
                if progress_bar is not None and "totalCount" in page:
                    total_pages = (page["totalCount"] + page_size - 1) // page_size
                    total_pages = min(total_pages, max_pages) if max_pages else total_pages
                    progress_bar.total = total_pages
                first_page = False

            all_results.extend(page.get(results_key, []))
            if progress_bar is not None:
                progress_bar.update(1)
                progress_bar.set_postfix({"items": len(all_results)})
    finally:
        if progress_bar is not None:
            progress_bar.close()

    return (all_results, metadata) if return_metadata else all_results

afetch_all_results_with_metadata async

afetch_all_results_with_metadata(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    show_progress=True,
)
Source code in pybdl/api/client.py
async def afetch_all_results_with_metadata(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    return await self.afetch_all_results(
        endpoint,
        method=method,
        params=params,
        headers=headers,
        results_key=results_key,
        page_size=page_size,
        max_pages=max_pages,
        return_metadata=True,
        show_progress=show_progress,
    )

afetch_single_result async

afetch_single_result(
    endpoint: str,
    *,
    method: str = "GET",
    results_key: Literal[None] = None,
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> dict[str, Any]
afetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> list[dict[str, Any]]
afetch_single_result(
    endpoint: str,
    *,
    return_metadata: Literal[True],
    results_key: Literal[None] = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]]
afetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    return_metadata: Literal[True],
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
afetch_single_result(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
    return_metadata=False,
)

Asynchronously fetch a single result, non-paginated.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
results_key str | None

If not None, extract this key from the JSON.

None
method str

HTTP method.

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
return_metadata bool

Also return metadata if True.

False

Returns:

Type Description
dict[str, Any] | list[dict[str, Any]] | tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Dictionary or list, optionally with separate metadata.

Source code in pybdl/api/client.py
async def afetch_single_result(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: bool = False,
) -> (
    dict[str, Any]
    | list[dict[str, Any]]
    | tuple[dict[str, Any], dict[str, Any]]
    | tuple[list[dict[str, Any]], dict[str, Any]]
):
    """
    Asynchronously fetch a single result, non-paginated.

    Args:
        endpoint: API endpoint.
        results_key: If not None, extract this key from the JSON.
        method: HTTP method.
        params: Query parameters.
        headers: Optional request headers.
        return_metadata: Also return metadata if True.

    Returns:
        Dictionary or list, optionally with separate metadata.
    """
    response = await self._request_async(
        endpoint=endpoint,
        method=method,
        params=params,
        headers=headers,
    )

    if results_key is None:
        return (response, {}) if return_metadata else response

    if not isinstance(response, dict) or results_key not in response:
        raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=response)

    results_val = cast(list[dict[str, Any]], response[results_key])
    if return_metadata:
        metadata = self._metadata_from_response(response, results_key)
        return results_val, metadata

    return results_val

afetch_single_result_with_metadata async

afetch_single_result_with_metadata(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
)
Source code in pybdl/api/client.py
async def afetch_single_result_with_metadata(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    return await self.afetch_single_result(
        endpoint,
        results_key=results_key,
        method=method,
        params=params,
        headers=headers,
        return_metadata=True,
    )

Data

data

DataAPI

DataAPI(config, extra_headers=None)

Bases: BaseAPIClient

Client for all BDL /data endpoints.

Provides Pythonic, paginated, and DataFrame-ready access to all public data endpoints in the Local Data Bank (BDL) API. Supports flexible parameterization, pagination, and format options for robust data retrieval.

Methods map directly to documented BDL endpoints under the /data namespace, enabling users to fetch statistical data by variable, unit, and locality.

Initialize base API client for BDL.

Parameters:

Name Type Description Default
config BDLConfig

BDL configuration object.

required
extra_headers dict[str, str] | None

Optional extra headers (e.g., Accept-Language) to include in requests.

None
Source code in pybdl/api/client.py
def __init__(self, config: BDLConfig, extra_headers: dict[str, str] | None = None):
    """
    Initialize base API client for BDL.

    Args:
        config: BDL configuration object.
        extra_headers: Optional extra headers (e.g., Accept-Language) to include in requests.
    """
    self.config = config
    is_registered = bool(config.api_key)
    quotas: QuotaMap = cast(
        QuotaMap,
        config.custom_quotas if config.custom_quotas is not None else DEFAULT_QUOTAS,
    )
    self._quota_cache = PersistentQuotaCache(
        config.quota_cache_enabled,
        cache_file=config.quota_cache_file,
        use_global_cache=config.use_global_cache,
    )
    self._sync_limiter = RateLimiter(
        quotas,
        is_registered,
        self._quota_cache,
        raise_on_limit=config.raise_on_rate_limit,
    )
    self._async_limiter = AsyncRateLimiter(
        quotas,
        is_registered,
        self._quota_cache,
        raise_on_limit=config.raise_on_rate_limit,
    )
    self._proxy_url = self._build_proxy_url()
    self._http_cache_path = resolve_http_cache_db_path(config.cache_backend, self._quota_cache.cache_file)
    default_headers = self._build_default_headers(extra_headers)
    self.session = build_sync_http_client(
        cache_backend=config.cache_backend,
        http_cache_db_path=self._http_cache_path,
        default_headers=default_headers,
        proxy=self._proxy_url,
    )
    self._async_client = build_async_http_client(
        cache_backend=config.cache_backend,
        http_cache_db_path=self._http_cache_path,
        default_headers=default_headers,
        proxy=self._proxy_url,
    )

config instance-attribute

config = config

session instance-attribute

session = build_sync_http_client(
    cache_backend=cache_backend,
    http_cache_db_path=_http_cache_path,
    default_headers=default_headers,
    proxy=_proxy_url,
)

get_data_by_variable

get_data_by_variable(
    variable_id,
    years=None,
    unit_parent_id=None,
    unit_level=None,
    aggregate_id=None,
    page=None,
    page_size=100,
    max_pages=None,
    format=None,
    lang=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
    return_metadata=False,
)
Source code in pybdl/api/data.py
def get_data_by_variable(
    self,
    variable_id: str,
    years: list[int] | None = None,
    unit_parent_id: str | None = None,
    unit_level: int | None = None,
    aggregate_id: int | None = None,
    page: int | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    format: FormatLiteral | None = None,
    lang: LanguageLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
    return_metadata: bool = False,
) -> _DataCollectionResult:
    endpoint = f"data/by-variable/{variable_id}"
    params, headers = self._prepare_collection_request(
        extra_params=self._data_by_variable_params(
            years,
            unit_parent_id,
            unit_level,
            aggregate_id,
            page,
            extra_query,
        ),
        format=format,
        lang=lang,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )
    return self._fetch_data_collection(
        endpoint,
        params=params,
        headers=headers,
        page_size=page_size,
        max_pages=max_pages,
        return_metadata=return_metadata,
    )

get_data_by_variable_with_metadata

get_data_by_variable_with_metadata(*args, **kwargs)

Retrieve data by variable and always return (results, metadata).

Source code in pybdl/api/data.py
def get_data_by_variable_with_metadata(
    self, *args: Any, **kwargs: Any
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    """Retrieve data by variable and always return `(results, metadata)`."""
    kwargs["return_metadata"] = True
    return cast(
        tuple[list[dict[str, Any]], dict[str, Any]],
        self.get_data_by_variable(*args, **kwargs),
    )

get_data_by_unit

get_data_by_unit(
    unit_id,
    variable_ids=None,
    *,
    variable_id=None,
    years=None,
    aggregate_id=None,
    page=None,
    page_size=100,
    format=None,
    lang=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
    return_metadata=False,
)
Source code in pybdl/api/data.py
def get_data_by_unit(
    self,
    unit_id: str,
    variable_ids: Sequence[str | int] | str | int | None = None,
    *,
    variable_id: Sequence[str | int] | str | int | None = None,
    years: list[int] | None = None,
    aggregate_id: int | None = None,
    page: int | None = None,
    page_size: int = 100,
    format: FormatLiteral | None = None,
    lang: LanguageLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
    return_metadata: bool = False,
) -> _DataCollectionResult:
    endpoint = f"data/by-unit/{unit_id}"
    params, headers = self._prepare_collection_request(
        extra_params=self._data_by_unit_params(variable_ids, variable_id, years, aggregate_id, page, extra_query),
        format=format,
        lang=lang,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )
    return self._fetch_single_page_data_collection(
        endpoint,
        params=params,
        headers=headers,
        page_size=page_size,
        return_metadata=return_metadata,
    )

get_data_by_unit_with_metadata

get_data_by_unit_with_metadata(*args, **kwargs)

Retrieve data by unit and always return (results, metadata).

Source code in pybdl/api/data.py
def get_data_by_unit_with_metadata(self, *args: Any, **kwargs: Any) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    """Retrieve data by unit and always return `(results, metadata)`."""
    kwargs["return_metadata"] = True
    return cast(
        tuple[list[dict[str, Any]], dict[str, Any]],
        self.get_data_by_unit(*args, **kwargs),
    )

get_data_by_variable_locality

get_data_by_variable_locality(
    variable_id,
    unit_parent_id,
    years=None,
    page=None,
    page_size=100,
    max_pages=None,
    format=None,
    lang=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
    return_metadata=False,
)
Source code in pybdl/api/data.py
def get_data_by_variable_locality(
    self,
    variable_id: str,
    unit_parent_id: str,
    years: list[int] | None = None,
    page: int | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    format: FormatLiteral | None = None,
    lang: LanguageLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
    return_metadata: bool = False,
) -> _DataCollectionResult:
    endpoint = f"data/localities/by-variable/{variable_id}"
    params, headers = self._prepare_collection_request(
        extra_params=self._data_by_variable_locality_params(unit_parent_id, years, page, extra_query),
        format=format,
        lang=lang,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )
    return self._fetch_data_collection(
        endpoint,
        params=params,
        headers=headers,
        page_size=page_size,
        max_pages=max_pages,
        return_metadata=return_metadata,
    )

get_data_by_variable_locality_with_metadata

get_data_by_variable_locality_with_metadata(
    *args, **kwargs
)

Retrieve locality data by variable and always return (results, metadata).

Source code in pybdl/api/data.py
def get_data_by_variable_locality_with_metadata(
    self, *args: Any, **kwargs: Any
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    """Retrieve locality data by variable and always return `(results, metadata)`."""
    kwargs["return_metadata"] = True
    return cast(
        tuple[list[dict[str, Any]], dict[str, Any]],
        self.get_data_by_variable_locality(*args, **kwargs),
    )

get_data_by_unit_locality

get_data_by_unit_locality(
    unit_id,
    variable_ids=None,
    *,
    variable_id=None,
    years=None,
    aggregate_id=None,
    page=None,
    page_size=100,
    max_pages=None,
    format=None,
    lang=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
    return_metadata=False,
)
Source code in pybdl/api/data.py
def get_data_by_unit_locality(
    self,
    unit_id: str,
    variable_ids: Sequence[str | int] | str | int | None = None,
    *,
    variable_id: Sequence[str | int] | str | int | None = None,
    years: list[int] | None = None,
    aggregate_id: int | None = None,
    page: int | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    format: FormatLiteral | None = None,
    lang: LanguageLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
    return_metadata: bool = False,
) -> _DataCollectionResult:
    endpoint = f"data/localities/by-unit/{unit_id}"
    params, headers = self._prepare_collection_request(
        extra_params=self._data_by_unit_locality_params(
            variable_ids,
            variable_id,
            years,
            aggregate_id,
            page,
            extra_query,
        ),
        format=format,
        lang=lang,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )
    return self._fetch_data_collection(
        endpoint,
        params=params,
        headers=headers,
        page_size=page_size,
        max_pages=max_pages,
        return_metadata=return_metadata,
    )

get_data_by_unit_locality_with_metadata

get_data_by_unit_locality_with_metadata(*args, **kwargs)

Retrieve locality data by unit and always return (results, metadata).

Source code in pybdl/api/data.py
def get_data_by_unit_locality_with_metadata(
    self, *args: Any, **kwargs: Any
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    """Retrieve locality data by unit and always return `(results, metadata)`."""
    kwargs["return_metadata"] = True
    return cast(
        tuple[list[dict[str, Any]], dict[str, Any]],
        self.get_data_by_unit_locality(*args, **kwargs),
    )

get_data_metadata

get_data_metadata(
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/data.py
def get_data_metadata(
    self,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return self._fetch_detail_endpoint(
        "data/metadata",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

aget_data_by_variable async

aget_data_by_variable(
    variable_id,
    years=None,
    unit_parent_id=None,
    unit_level=None,
    aggregate_id=None,
    page=None,
    page_size=100,
    max_pages=None,
    format=None,
    lang=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
    return_metadata=False,
)
Source code in pybdl/api/data.py
async def aget_data_by_variable(
    self,
    variable_id: str,
    years: list[int] | None = None,
    unit_parent_id: str | None = None,
    unit_level: int | None = None,
    aggregate_id: int | None = None,
    page: int | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    format: FormatLiteral | None = None,
    lang: LanguageLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
    return_metadata: bool = False,
) -> _DataCollectionResult:
    endpoint = f"data/by-variable/{variable_id}"
    params, headers = self._prepare_collection_request(
        extra_params=self._data_by_variable_params(
            years,
            unit_parent_id,
            unit_level,
            aggregate_id,
            page,
            extra_query,
        ),
        format=format,
        lang=lang,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )
    return await self._afetch_data_collection(
        endpoint,
        params=params,
        headers=headers,
        page_size=page_size,
        max_pages=max_pages,
        return_metadata=return_metadata,
    )

aget_data_by_variable_with_metadata async

aget_data_by_variable_with_metadata(*args, **kwargs)

Asynchronously retrieve data by variable and always return (results, metadata).

Source code in pybdl/api/data.py
async def aget_data_by_variable_with_metadata(
    self, *args: Any, **kwargs: Any
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    """Asynchronously retrieve data by variable and always return `(results, metadata)`."""
    kwargs["return_metadata"] = True
    return cast(
        tuple[list[dict[str, Any]], dict[str, Any]],
        await self.aget_data_by_variable(*args, **kwargs),
    )

aget_data_by_unit async

aget_data_by_unit(
    unit_id,
    variable_ids=None,
    *,
    variable_id=None,
    years=None,
    aggregate_id=None,
    page=None,
    page_size=100,
    format=None,
    lang=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
    return_metadata=False,
)
Source code in pybdl/api/data.py
async def aget_data_by_unit(
    self,
    unit_id: str,
    variable_ids: Sequence[str | int] | str | int | None = None,
    *,
    variable_id: Sequence[str | int] | str | int | None = None,
    years: list[int] | None = None,
    aggregate_id: int | None = None,
    page: int | None = None,
    page_size: int = 100,
    format: FormatLiteral | None = None,
    lang: LanguageLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
    return_metadata: bool = False,
) -> _DataCollectionResult:
    endpoint = f"data/by-unit/{unit_id}"
    params, headers = self._prepare_collection_request(
        extra_params=self._data_by_unit_params(variable_ids, variable_id, years, aggregate_id, page, extra_query),
        format=format,
        lang=lang,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )
    return await self._afetch_single_page_data_collection(
        endpoint,
        params=params,
        headers=headers,
        page_size=page_size,
        return_metadata=return_metadata,
    )

aget_data_by_unit_with_metadata async

aget_data_by_unit_with_metadata(*args, **kwargs)

Asynchronously retrieve data by unit and always return (results, metadata).

Source code in pybdl/api/data.py
async def aget_data_by_unit_with_metadata(
    self, *args: Any, **kwargs: Any
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    """Asynchronously retrieve data by unit and always return `(results, metadata)`."""
    kwargs["return_metadata"] = True
    return cast(
        tuple[list[dict[str, Any]], dict[str, Any]],
        await self.aget_data_by_unit(*args, **kwargs),
    )

aget_data_by_variable_locality async

aget_data_by_variable_locality(
    variable_id,
    unit_parent_id,
    years=None,
    page=None,
    page_size=100,
    max_pages=None,
    format=None,
    lang=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
    return_metadata=False,
)
Source code in pybdl/api/data.py
async def aget_data_by_variable_locality(
    self,
    variable_id: str,
    unit_parent_id: str,
    years: list[int] | None = None,
    page: int | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    format: FormatLiteral | None = None,
    lang: LanguageLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
    return_metadata: bool = False,
) -> _DataCollectionResult:
    endpoint = f"data/localities/by-variable/{variable_id}"
    params, headers = self._prepare_collection_request(
        extra_params=self._data_by_variable_locality_params(unit_parent_id, years, page, extra_query),
        format=format,
        lang=lang,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )
    return await self._afetch_data_collection(
        endpoint,
        params=params,
        headers=headers,
        page_size=page_size,
        max_pages=max_pages,
        return_metadata=return_metadata,
    )

aget_data_by_variable_locality_with_metadata async

aget_data_by_variable_locality_with_metadata(
    *args, **kwargs
)

Asynchronously retrieve locality data by variable and always return (results, metadata).

Source code in pybdl/api/data.py
async def aget_data_by_variable_locality_with_metadata(
    self, *args: Any, **kwargs: Any
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    """Asynchronously retrieve locality data by variable and always return `(results, metadata)`."""
    kwargs["return_metadata"] = True
    return cast(
        tuple[list[dict[str, Any]], dict[str, Any]],
        await self.aget_data_by_variable_locality(*args, **kwargs),
    )

aget_data_by_unit_locality async

aget_data_by_unit_locality(
    unit_id,
    variable_ids=None,
    *,
    variable_id=None,
    years=None,
    aggregate_id=None,
    page=None,
    page_size=100,
    max_pages=None,
    format=None,
    lang=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
    return_metadata=False,
)
Source code in pybdl/api/data.py
async def aget_data_by_unit_locality(
    self,
    unit_id: str,
    variable_ids: Sequence[str | int] | str | int | None = None,
    *,
    variable_id: Sequence[str | int] | str | int | None = None,
    years: list[int] | None = None,
    aggregate_id: int | None = None,
    page: int | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    format: FormatLiteral | None = None,
    lang: LanguageLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
    return_metadata: bool = False,
) -> _DataCollectionResult:
    endpoint = f"data/localities/by-unit/{unit_id}"
    params, headers = self._prepare_collection_request(
        extra_params=self._data_by_unit_locality_params(
            variable_ids,
            variable_id,
            years,
            aggregate_id,
            page,
            extra_query,
        ),
        format=format,
        lang=lang,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )
    return await self._afetch_data_collection(
        endpoint,
        params=params,
        headers=headers,
        page_size=page_size,
        max_pages=max_pages,
        return_metadata=return_metadata,
    )

aget_data_by_unit_locality_with_metadata async

aget_data_by_unit_locality_with_metadata(*args, **kwargs)

Asynchronously retrieve locality data by unit and always return (results, metadata).

Source code in pybdl/api/data.py
async def aget_data_by_unit_locality_with_metadata(
    self, *args: Any, **kwargs: Any
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    """Asynchronously retrieve locality data by unit and always return `(results, metadata)`."""
    kwargs["return_metadata"] = True
    return cast(
        tuple[list[dict[str, Any]], dict[str, Any]],
        await self.aget_data_by_unit_locality(*args, **kwargs),
    )

aget_data_metadata async

aget_data_metadata(
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/data.py
async def aget_data_metadata(
    self,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return await self._afetch_detail_endpoint(
        "data/metadata",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

close

close()

Close synchronous HTTP resources.

Source code in pybdl/api/client.py
def close(self) -> None:
    """Close synchronous HTTP resources."""
    self.session.close()

aclose async

aclose()

Close synchronous and asynchronous HTTP resources.

Source code in pybdl/api/client.py
async def aclose(self) -> None:
    """Close synchronous and asynchronous HTTP resources."""
    self.close()
    await self._async_client.aclose()

__enter__

__enter__()
Source code in pybdl/api/client.py
def __enter__(self) -> "BaseAPIClient":
    return self

__exit__

__exit__(exc_type, exc_val, exc_tb)
Source code in pybdl/api/client.py
def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> Literal[False]:
    self.close()
    return False

__aenter__ async

__aenter__()
Source code in pybdl/api/client.py
async def __aenter__(self) -> "BaseAPIClient":
    return self

__aexit__ async

__aexit__(exc_type, exc_val, exc_tb)
Source code in pybdl/api/client.py
async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> Literal[False]:
    await self.aclose()
    return False

fetch_all_results

fetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[False] = False,
    show_progress: bool = True,
) -> list[dict[str, Any]]
fetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[True],
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
fetch_all_results(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    return_metadata=False,
    show_progress=True,
)

Fetch paginated results synchronously and combine them into a single list.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
method str

HTTP method (default: GET).

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
results_key str

Key for extracting data from each page.

'results'
page_size int

Items per page.

100
max_pages int | None

Optional limit of pages.

None
return_metadata bool

If True, return (results, metadata).

False
show_progress bool

Display progress via tqdm.

True

Returns:

Type Description
list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Combined list of results, optionally with metadata.

Source code in pybdl/api/client.py
def fetch_all_results(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: bool = False,
    show_progress: bool = True,
) -> list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    """
    Fetch paginated results synchronously and combine them into a single list.

    Args:
        endpoint: API endpoint.
        method: HTTP method (default: GET).
        params: Query parameters.
        headers: Optional request headers.
        results_key: Key for extracting data from each page.
        page_size: Items per page.
        max_pages: Optional limit of pages.
        return_metadata: If True, return (results, metadata).
        show_progress: Display progress via tqdm.

    Returns:
        Combined list of results, optionally with metadata.
    """
    all_results: list[dict[str, Any]] = []
    metadata: dict[str, Any] = {}
    progress_bar = (
        tqdm(desc=f"Fetching {endpoint.split('/')[-1]}", unit=" pages", leave=True) if show_progress else None
    )

    first_page = True
    try:
        for page in self._paginated_request_sync(
            endpoint,
            method=method,
            params=params,
            headers=headers,
            results_key=results_key,
            page_size=page_size,
            max_pages=max_pages,
        ):
            if results_key not in page:
                raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=page)
            if first_page and return_metadata:
                metadata = self._metadata_from_response(page, results_key)
                if progress_bar is not None and "totalCount" in page:
                    total_pages = (page["totalCount"] + page_size - 1) // page_size
                    total_pages = min(total_pages, max_pages) if max_pages else total_pages
                    progress_bar.total = total_pages
                first_page = False

            all_results.extend(page.get(results_key, []))

            if progress_bar is not None:
                progress_bar.update(1)
                progress_bar.set_postfix({"items": len(all_results)})
    finally:
        if progress_bar is not None:
            progress_bar.close()

    return (all_results, metadata) if return_metadata else all_results

fetch_all_results_with_metadata

fetch_all_results_with_metadata(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    show_progress=True,
)
Source code in pybdl/api/client.py
def fetch_all_results_with_metadata(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    return self.fetch_all_results(
        endpoint,
        method=method,
        params=params,
        headers=headers,
        results_key=results_key,
        page_size=page_size,
        max_pages=max_pages,
        return_metadata=True,
        show_progress=show_progress,
    )

fetch_single_result

fetch_single_result(
    endpoint: str,
    *,
    results_key: None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> dict[str, Any]
fetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> list[dict[str, Any]]
fetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[True],
) -> tuple[list[dict[str, Any]], dict[str, Any]]
fetch_single_result(
    endpoint: str,
    *,
    results_key: None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[True],
) -> tuple[dict[str, Any], dict[str, Any]]
fetch_single_result(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
    return_metadata=False,
)

Fetch a single result, non-paginated (sync).

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
results_key str | None

If not None, extract this key from the JSON.

None
method str

HTTP method.

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
return_metadata bool

Also return metadata if True.

False

Returns:

Type Description
dict[str, Any] | list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]] | tuple[dict[str, Any], dict[str, Any]]

Dictionary or list, optionally with separate metadata.

Source code in pybdl/api/client.py
def fetch_single_result(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: bool = False,
) -> (
    dict[str, Any]
    | list[dict[str, Any]]
    | tuple[list[dict[str, Any]], dict[str, Any]]
    | tuple[dict[str, Any], dict[str, Any]]
):
    """
    Fetch a single result, non-paginated (sync).

    Args:
        endpoint: API endpoint.
        results_key: If not None, extract this key from the JSON.
        method: HTTP method.
        params: Query parameters.
        headers: Optional request headers.
        return_metadata: Also return metadata if True.

    Returns:
        Dictionary or list, optionally with separate metadata.
    """
    response = self._request_sync(
        endpoint=endpoint,
        method=method,
        params=params,
        headers=headers,
    )

    if results_key is None:
        if return_metadata:
            return response, {}
        return response

    if not isinstance(response, dict) or results_key not in response:
        raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=response)

    results_val = response[results_key]
    if return_metadata:
        metadata = self._metadata_from_response(response, results_key)
        return results_val, metadata

    return results_val

fetch_single_result_with_metadata

fetch_single_result_with_metadata(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
)
Source code in pybdl/api/client.py
def fetch_single_result_with_metadata(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    return self.fetch_single_result(
        endpoint,
        results_key=results_key,
        method=method,
        params=params,
        headers=headers,
        return_metadata=True,
    )

afetch_all_results async

afetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[False] = False,
    show_progress: bool = True,
) -> list[dict[str, Any]]
afetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[True],
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
afetch_all_results(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    return_metadata=False,
    show_progress=True,
)

Asynchronously fetch paginated results and combine them into a single list.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
method str

HTTP method (default: GET).

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
results_key str

Key for extracting data from each page.

'results'
page_size int

Items per page.

100
max_pages int | None

Optional limit of pages.

None
return_metadata bool

If True, return (results, metadata).

False
show_progress bool

Display progress via tqdm.

True

Returns:

Type Description
list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Combined list of results, optionally with metadata.

Source code in pybdl/api/client.py
async def afetch_all_results(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: bool = False,
    show_progress: bool = True,
) -> list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    """
    Asynchronously fetch paginated results and combine them into a single list.

    Args:
        endpoint: API endpoint.
        method: HTTP method (default: GET).
        params: Query parameters.
        headers: Optional request headers.
        results_key: Key for extracting data from each page.
        page_size: Items per page.
        max_pages: Optional limit of pages.
        return_metadata: If True, return (results, metadata).
        show_progress: Display progress via tqdm.

    Returns:
        Combined list of results, optionally with metadata.
    """
    all_results: list[dict[str, Any]] = []
    metadata: dict[str, Any] = {}
    first_page = True
    progress_bar = (
        tqdm(desc=f"Fetching {endpoint.split('/')[-1]} (async)", unit=" pages", leave=True)
        if show_progress
        else None
    )

    try:
        async for page in self._paginated_request_async(
            endpoint,
            method=method,
            params=params,
            headers=headers,
            results_key=results_key,
            page_size=page_size,
            max_pages=max_pages,
        ):
            if results_key not in page:
                raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=page)
            if first_page and return_metadata:
                metadata = self._metadata_from_response(page, results_key)
                if progress_bar is not None and "totalCount" in page:
                    total_pages = (page["totalCount"] + page_size - 1) // page_size
                    total_pages = min(total_pages, max_pages) if max_pages else total_pages
                    progress_bar.total = total_pages
                first_page = False

            all_results.extend(page.get(results_key, []))
            if progress_bar is not None:
                progress_bar.update(1)
                progress_bar.set_postfix({"items": len(all_results)})
    finally:
        if progress_bar is not None:
            progress_bar.close()

    return (all_results, metadata) if return_metadata else all_results

afetch_all_results_with_metadata async

afetch_all_results_with_metadata(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    show_progress=True,
)
Source code in pybdl/api/client.py
async def afetch_all_results_with_metadata(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    return await self.afetch_all_results(
        endpoint,
        method=method,
        params=params,
        headers=headers,
        results_key=results_key,
        page_size=page_size,
        max_pages=max_pages,
        return_metadata=True,
        show_progress=show_progress,
    )

afetch_single_result async

afetch_single_result(
    endpoint: str,
    *,
    method: str = "GET",
    results_key: Literal[None] = None,
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> dict[str, Any]
afetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> list[dict[str, Any]]
afetch_single_result(
    endpoint: str,
    *,
    return_metadata: Literal[True],
    results_key: Literal[None] = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]]
afetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    return_metadata: Literal[True],
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
afetch_single_result(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
    return_metadata=False,
)

Asynchronously fetch a single result, non-paginated.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
results_key str | None

If not None, extract this key from the JSON.

None
method str

HTTP method.

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
return_metadata bool

Also return metadata if True.

False

Returns:

Type Description
dict[str, Any] | list[dict[str, Any]] | tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Dictionary or list, optionally with separate metadata.

Source code in pybdl/api/client.py
async def afetch_single_result(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: bool = False,
) -> (
    dict[str, Any]
    | list[dict[str, Any]]
    | tuple[dict[str, Any], dict[str, Any]]
    | tuple[list[dict[str, Any]], dict[str, Any]]
):
    """
    Asynchronously fetch a single result, non-paginated.

    Args:
        endpoint: API endpoint.
        results_key: If not None, extract this key from the JSON.
        method: HTTP method.
        params: Query parameters.
        headers: Optional request headers.
        return_metadata: Also return metadata if True.

    Returns:
        Dictionary or list, optionally with separate metadata.
    """
    response = await self._request_async(
        endpoint=endpoint,
        method=method,
        params=params,
        headers=headers,
    )

    if results_key is None:
        return (response, {}) if return_metadata else response

    if not isinstance(response, dict) or results_key not in response:
        raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=response)

    results_val = cast(list[dict[str, Any]], response[results_key])
    if return_metadata:
        metadata = self._metadata_from_response(response, results_key)
        return results_val, metadata

    return results_val

afetch_single_result_with_metadata async

afetch_single_result_with_metadata(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
)
Source code in pybdl/api/client.py
async def afetch_single_result_with_metadata(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    return await self.afetch_single_result(
        endpoint,
        results_key=results_key,
        method=method,
        params=params,
        headers=headers,
        return_metadata=True,
    )

Levels

levels

LevelsAPI

LevelsAPI(config, extra_headers=None)

Bases: BaseAPIClient

Client for the BDL /levels endpoints.

Initialize base API client for BDL.

Parameters:

Name Type Description Default
config BDLConfig

BDL configuration object.

required
extra_headers dict[str, str] | None

Optional extra headers (e.g., Accept-Language) to include in requests.

None
Source code in pybdl/api/client.py
def __init__(self, config: BDLConfig, extra_headers: dict[str, str] | None = None):
    """
    Initialize base API client for BDL.

    Args:
        config: BDL configuration object.
        extra_headers: Optional extra headers (e.g., Accept-Language) to include in requests.
    """
    self.config = config
    is_registered = bool(config.api_key)
    quotas: QuotaMap = cast(
        QuotaMap,
        config.custom_quotas if config.custom_quotas is not None else DEFAULT_QUOTAS,
    )
    self._quota_cache = PersistentQuotaCache(
        config.quota_cache_enabled,
        cache_file=config.quota_cache_file,
        use_global_cache=config.use_global_cache,
    )
    self._sync_limiter = RateLimiter(
        quotas,
        is_registered,
        self._quota_cache,
        raise_on_limit=config.raise_on_rate_limit,
    )
    self._async_limiter = AsyncRateLimiter(
        quotas,
        is_registered,
        self._quota_cache,
        raise_on_limit=config.raise_on_rate_limit,
    )
    self._proxy_url = self._build_proxy_url()
    self._http_cache_path = resolve_http_cache_db_path(config.cache_backend, self._quota_cache.cache_file)
    default_headers = self._build_default_headers(extra_headers)
    self.session = build_sync_http_client(
        cache_backend=config.cache_backend,
        http_cache_db_path=self._http_cache_path,
        default_headers=default_headers,
        proxy=self._proxy_url,
    )
    self._async_client = build_async_http_client(
        cache_backend=config.cache_backend,
        http_cache_db_path=self._http_cache_path,
        default_headers=default_headers,
        proxy=self._proxy_url,
    )

config instance-attribute

config = config

session instance-attribute

session = build_sync_http_client(
    cache_backend=cache_backend,
    http_cache_db_path=_http_cache_path,
    default_headers=default_headers,
    proxy=_proxy_url,
)

list_levels

list_levels(
    sort=None,
    page_size=100,
    max_pages=None,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/levels.py
def list_levels(
    self,
    sort: str | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    return self._fetch_collection_endpoint(
        "levels",
        extra_params=self._list_params(sort, extra_query),
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
        page_size=page_size,
        max_pages=max_pages,
    )

get_level

get_level(
    level_id,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/levels.py
def get_level(
    self,
    level_id: int,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return self._fetch_detail_endpoint(
        f"levels/{level_id}",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

get_levels_metadata

get_levels_metadata(
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/levels.py
def get_levels_metadata(
    self,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return self._fetch_detail_endpoint(
        "levels/metadata",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

alist_levels async

alist_levels(
    sort=None,
    page_size=100,
    max_pages=None,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/levels.py
async def alist_levels(
    self,
    sort: str | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    return await self._afetch_collection_endpoint(
        "levels",
        extra_params=self._list_params(sort, extra_query),
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
        page_size=page_size,
        max_pages=max_pages,
    )

aget_level async

aget_level(
    level_id,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/levels.py
async def aget_level(
    self,
    level_id: int,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return await self._afetch_detail_endpoint(
        f"levels/{level_id}",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

aget_levels_metadata async

aget_levels_metadata(
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/levels.py
async def aget_levels_metadata(
    self,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return await self._afetch_detail_endpoint(
        "levels/metadata",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

close

close()

Close synchronous HTTP resources.

Source code in pybdl/api/client.py
def close(self) -> None:
    """Close synchronous HTTP resources."""
    self.session.close()

aclose async

aclose()

Close synchronous and asynchronous HTTP resources.

Source code in pybdl/api/client.py
async def aclose(self) -> None:
    """Close synchronous and asynchronous HTTP resources."""
    self.close()
    await self._async_client.aclose()

__enter__

__enter__()
Source code in pybdl/api/client.py
def __enter__(self) -> "BaseAPIClient":
    return self

__exit__

__exit__(exc_type, exc_val, exc_tb)
Source code in pybdl/api/client.py
def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> Literal[False]:
    self.close()
    return False

__aenter__ async

__aenter__()
Source code in pybdl/api/client.py
async def __aenter__(self) -> "BaseAPIClient":
    return self

__aexit__ async

__aexit__(exc_type, exc_val, exc_tb)
Source code in pybdl/api/client.py
async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> Literal[False]:
    await self.aclose()
    return False

fetch_all_results

fetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[False] = False,
    show_progress: bool = True,
) -> list[dict[str, Any]]
fetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[True],
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
fetch_all_results(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    return_metadata=False,
    show_progress=True,
)

Fetch paginated results synchronously and combine them into a single list.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
method str

HTTP method (default: GET).

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
results_key str

Key for extracting data from each page.

'results'
page_size int

Items per page.

100
max_pages int | None

Optional limit of pages.

None
return_metadata bool

If True, return (results, metadata).

False
show_progress bool

Display progress via tqdm.

True

Returns:

Type Description
list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Combined list of results, optionally with metadata.

Source code in pybdl/api/client.py
def fetch_all_results(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: bool = False,
    show_progress: bool = True,
) -> list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    """
    Fetch paginated results synchronously and combine them into a single list.

    Args:
        endpoint: API endpoint.
        method: HTTP method (default: GET).
        params: Query parameters.
        headers: Optional request headers.
        results_key: Key for extracting data from each page.
        page_size: Items per page.
        max_pages: Optional limit of pages.
        return_metadata: If True, return (results, metadata).
        show_progress: Display progress via tqdm.

    Returns:
        Combined list of results, optionally with metadata.
    """
    all_results: list[dict[str, Any]] = []
    metadata: dict[str, Any] = {}
    progress_bar = (
        tqdm(desc=f"Fetching {endpoint.split('/')[-1]}", unit=" pages", leave=True) if show_progress else None
    )

    first_page = True
    try:
        for page in self._paginated_request_sync(
            endpoint,
            method=method,
            params=params,
            headers=headers,
            results_key=results_key,
            page_size=page_size,
            max_pages=max_pages,
        ):
            if results_key not in page:
                raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=page)
            if first_page and return_metadata:
                metadata = self._metadata_from_response(page, results_key)
                if progress_bar is not None and "totalCount" in page:
                    total_pages = (page["totalCount"] + page_size - 1) // page_size
                    total_pages = min(total_pages, max_pages) if max_pages else total_pages
                    progress_bar.total = total_pages
                first_page = False

            all_results.extend(page.get(results_key, []))

            if progress_bar is not None:
                progress_bar.update(1)
                progress_bar.set_postfix({"items": len(all_results)})
    finally:
        if progress_bar is not None:
            progress_bar.close()

    return (all_results, metadata) if return_metadata else all_results

fetch_all_results_with_metadata

fetch_all_results_with_metadata(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    show_progress=True,
)
Source code in pybdl/api/client.py
def fetch_all_results_with_metadata(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    return self.fetch_all_results(
        endpoint,
        method=method,
        params=params,
        headers=headers,
        results_key=results_key,
        page_size=page_size,
        max_pages=max_pages,
        return_metadata=True,
        show_progress=show_progress,
    )

fetch_single_result

fetch_single_result(
    endpoint: str,
    *,
    results_key: None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> dict[str, Any]
fetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> list[dict[str, Any]]
fetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[True],
) -> tuple[list[dict[str, Any]], dict[str, Any]]
fetch_single_result(
    endpoint: str,
    *,
    results_key: None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[True],
) -> tuple[dict[str, Any], dict[str, Any]]
fetch_single_result(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
    return_metadata=False,
)

Fetch a single result, non-paginated (sync).

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
results_key str | None

If not None, extract this key from the JSON.

None
method str

HTTP method.

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
return_metadata bool

Also return metadata if True.

False

Returns:

Type Description
dict[str, Any] | list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]] | tuple[dict[str, Any], dict[str, Any]]

Dictionary or list, optionally with separate metadata.

Source code in pybdl/api/client.py
def fetch_single_result(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: bool = False,
) -> (
    dict[str, Any]
    | list[dict[str, Any]]
    | tuple[list[dict[str, Any]], dict[str, Any]]
    | tuple[dict[str, Any], dict[str, Any]]
):
    """
    Fetch a single result, non-paginated (sync).

    Args:
        endpoint: API endpoint.
        results_key: If not None, extract this key from the JSON.
        method: HTTP method.
        params: Query parameters.
        headers: Optional request headers.
        return_metadata: Also return metadata if True.

    Returns:
        Dictionary or list, optionally with separate metadata.
    """
    response = self._request_sync(
        endpoint=endpoint,
        method=method,
        params=params,
        headers=headers,
    )

    if results_key is None:
        if return_metadata:
            return response, {}
        return response

    if not isinstance(response, dict) or results_key not in response:
        raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=response)

    results_val = response[results_key]
    if return_metadata:
        metadata = self._metadata_from_response(response, results_key)
        return results_val, metadata

    return results_val

fetch_single_result_with_metadata

fetch_single_result_with_metadata(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
)
Source code in pybdl/api/client.py
def fetch_single_result_with_metadata(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    return self.fetch_single_result(
        endpoint,
        results_key=results_key,
        method=method,
        params=params,
        headers=headers,
        return_metadata=True,
    )

afetch_all_results async

afetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[False] = False,
    show_progress: bool = True,
) -> list[dict[str, Any]]
afetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[True],
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
afetch_all_results(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    return_metadata=False,
    show_progress=True,
)

Asynchronously fetch paginated results and combine them into a single list.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
method str

HTTP method (default: GET).

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
results_key str

Key for extracting data from each page.

'results'
page_size int

Items per page.

100
max_pages int | None

Optional limit of pages.

None
return_metadata bool

If True, return (results, metadata).

False
show_progress bool

Display progress via tqdm.

True

Returns:

Type Description
list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Combined list of results, optionally with metadata.

Source code in pybdl/api/client.py
async def afetch_all_results(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: bool = False,
    show_progress: bool = True,
) -> list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    """
    Asynchronously fetch paginated results and combine them into a single list.

    Args:
        endpoint: API endpoint.
        method: HTTP method (default: GET).
        params: Query parameters.
        headers: Optional request headers.
        results_key: Key for extracting data from each page.
        page_size: Items per page.
        max_pages: Optional limit of pages.
        return_metadata: If True, return (results, metadata).
        show_progress: Display progress via tqdm.

    Returns:
        Combined list of results, optionally with metadata.
    """
    all_results: list[dict[str, Any]] = []
    metadata: dict[str, Any] = {}
    first_page = True
    progress_bar = (
        tqdm(desc=f"Fetching {endpoint.split('/')[-1]} (async)", unit=" pages", leave=True)
        if show_progress
        else None
    )

    try:
        async for page in self._paginated_request_async(
            endpoint,
            method=method,
            params=params,
            headers=headers,
            results_key=results_key,
            page_size=page_size,
            max_pages=max_pages,
        ):
            if results_key not in page:
                raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=page)
            if first_page and return_metadata:
                metadata = self._metadata_from_response(page, results_key)
                if progress_bar is not None and "totalCount" in page:
                    total_pages = (page["totalCount"] + page_size - 1) // page_size
                    total_pages = min(total_pages, max_pages) if max_pages else total_pages
                    progress_bar.total = total_pages
                first_page = False

            all_results.extend(page.get(results_key, []))
            if progress_bar is not None:
                progress_bar.update(1)
                progress_bar.set_postfix({"items": len(all_results)})
    finally:
        if progress_bar is not None:
            progress_bar.close()

    return (all_results, metadata) if return_metadata else all_results

afetch_all_results_with_metadata async

afetch_all_results_with_metadata(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    show_progress=True,
)
Source code in pybdl/api/client.py
async def afetch_all_results_with_metadata(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    return await self.afetch_all_results(
        endpoint,
        method=method,
        params=params,
        headers=headers,
        results_key=results_key,
        page_size=page_size,
        max_pages=max_pages,
        return_metadata=True,
        show_progress=show_progress,
    )

afetch_single_result async

afetch_single_result(
    endpoint: str,
    *,
    method: str = "GET",
    results_key: Literal[None] = None,
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> dict[str, Any]
afetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> list[dict[str, Any]]
afetch_single_result(
    endpoint: str,
    *,
    return_metadata: Literal[True],
    results_key: Literal[None] = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]]
afetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    return_metadata: Literal[True],
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
afetch_single_result(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
    return_metadata=False,
)

Asynchronously fetch a single result, non-paginated.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
results_key str | None

If not None, extract this key from the JSON.

None
method str

HTTP method.

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
return_metadata bool

Also return metadata if True.

False

Returns:

Type Description
dict[str, Any] | list[dict[str, Any]] | tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Dictionary or list, optionally with separate metadata.

Source code in pybdl/api/client.py
async def afetch_single_result(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: bool = False,
) -> (
    dict[str, Any]
    | list[dict[str, Any]]
    | tuple[dict[str, Any], dict[str, Any]]
    | tuple[list[dict[str, Any]], dict[str, Any]]
):
    """
    Asynchronously fetch a single result, non-paginated.

    Args:
        endpoint: API endpoint.
        results_key: If not None, extract this key from the JSON.
        method: HTTP method.
        params: Query parameters.
        headers: Optional request headers.
        return_metadata: Also return metadata if True.

    Returns:
        Dictionary or list, optionally with separate metadata.
    """
    response = await self._request_async(
        endpoint=endpoint,
        method=method,
        params=params,
        headers=headers,
    )

    if results_key is None:
        return (response, {}) if return_metadata else response

    if not isinstance(response, dict) or results_key not in response:
        raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=response)

    results_val = cast(list[dict[str, Any]], response[results_key])
    if return_metadata:
        metadata = self._metadata_from_response(response, results_key)
        return results_val, metadata

    return results_val

afetch_single_result_with_metadata async

afetch_single_result_with_metadata(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
)
Source code in pybdl/api/client.py
async def afetch_single_result_with_metadata(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    return await self.afetch_single_result(
        endpoint,
        results_key=results_key,
        method=method,
        params=params,
        headers=headers,
        return_metadata=True,
    )

Measures

measures

MeasuresAPI

MeasuresAPI(config, extra_headers=None)

Bases: BaseAPIClient

Client for the BDL /measures endpoints.

Initialize base API client for BDL.

Parameters:

Name Type Description Default
config BDLConfig

BDL configuration object.

required
extra_headers dict[str, str] | None

Optional extra headers (e.g., Accept-Language) to include in requests.

None
Source code in pybdl/api/client.py
def __init__(self, config: BDLConfig, extra_headers: dict[str, str] | None = None):
    """
    Initialize base API client for BDL.

    Args:
        config: BDL configuration object.
        extra_headers: Optional extra headers (e.g., Accept-Language) to include in requests.
    """
    self.config = config
    is_registered = bool(config.api_key)
    quotas: QuotaMap = cast(
        QuotaMap,
        config.custom_quotas if config.custom_quotas is not None else DEFAULT_QUOTAS,
    )
    self._quota_cache = PersistentQuotaCache(
        config.quota_cache_enabled,
        cache_file=config.quota_cache_file,
        use_global_cache=config.use_global_cache,
    )
    self._sync_limiter = RateLimiter(
        quotas,
        is_registered,
        self._quota_cache,
        raise_on_limit=config.raise_on_rate_limit,
    )
    self._async_limiter = AsyncRateLimiter(
        quotas,
        is_registered,
        self._quota_cache,
        raise_on_limit=config.raise_on_rate_limit,
    )
    self._proxy_url = self._build_proxy_url()
    self._http_cache_path = resolve_http_cache_db_path(config.cache_backend, self._quota_cache.cache_file)
    default_headers = self._build_default_headers(extra_headers)
    self.session = build_sync_http_client(
        cache_backend=config.cache_backend,
        http_cache_db_path=self._http_cache_path,
        default_headers=default_headers,
        proxy=self._proxy_url,
    )
    self._async_client = build_async_http_client(
        cache_backend=config.cache_backend,
        http_cache_db_path=self._http_cache_path,
        default_headers=default_headers,
        proxy=self._proxy_url,
    )

config instance-attribute

config = config

session instance-attribute

session = build_sync_http_client(
    cache_backend=cache_backend,
    http_cache_db_path=_http_cache_path,
    default_headers=default_headers,
    proxy=_proxy_url,
)

list_measures

list_measures(
    sort=None,
    page_size=100,
    max_pages=None,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/measures.py
def list_measures(
    self,
    sort: str | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    return self._fetch_collection_endpoint(
        "measures",
        extra_params=self._list_params(sort, extra_query),
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
        page_size=page_size,
        max_pages=max_pages,
    )

get_measure

get_measure(
    measure_id,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/measures.py
def get_measure(
    self,
    measure_id: int,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return self._fetch_detail_endpoint(
        f"measures/{measure_id}",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

get_measures_metadata

get_measures_metadata(
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/measures.py
def get_measures_metadata(
    self,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return self._fetch_detail_endpoint(
        "measures/metadata",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

alist_measures async

alist_measures(
    sort=None,
    page_size=100,
    max_pages=None,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/measures.py
async def alist_measures(
    self,
    sort: str | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    return await self._afetch_collection_endpoint(
        "measures",
        extra_params=self._list_params(sort, extra_query),
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
        page_size=page_size,
        max_pages=max_pages,
    )

aget_measure async

aget_measure(
    measure_id,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/measures.py
async def aget_measure(
    self,
    measure_id: int,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return await self._afetch_detail_endpoint(
        f"measures/{measure_id}",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

aget_measures_metadata async

aget_measures_metadata(
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/measures.py
async def aget_measures_metadata(
    self,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return await self._afetch_detail_endpoint(
        "measures/metadata",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

close

close()

Close synchronous HTTP resources.

Source code in pybdl/api/client.py
def close(self) -> None:
    """Close synchronous HTTP resources."""
    self.session.close()

aclose async

aclose()

Close synchronous and asynchronous HTTP resources.

Source code in pybdl/api/client.py
async def aclose(self) -> None:
    """Close synchronous and asynchronous HTTP resources."""
    self.close()
    await self._async_client.aclose()

__enter__

__enter__()
Source code in pybdl/api/client.py
def __enter__(self) -> "BaseAPIClient":
    return self

__exit__

__exit__(exc_type, exc_val, exc_tb)
Source code in pybdl/api/client.py
def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> Literal[False]:
    self.close()
    return False

__aenter__ async

__aenter__()
Source code in pybdl/api/client.py
async def __aenter__(self) -> "BaseAPIClient":
    return self

__aexit__ async

__aexit__(exc_type, exc_val, exc_tb)
Source code in pybdl/api/client.py
async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> Literal[False]:
    await self.aclose()
    return False

fetch_all_results

fetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[False] = False,
    show_progress: bool = True,
) -> list[dict[str, Any]]
fetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[True],
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
fetch_all_results(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    return_metadata=False,
    show_progress=True,
)

Fetch paginated results synchronously and combine them into a single list.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
method str

HTTP method (default: GET).

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
results_key str

Key for extracting data from each page.

'results'
page_size int

Items per page.

100
max_pages int | None

Optional limit of pages.

None
return_metadata bool

If True, return (results, metadata).

False
show_progress bool

Display progress via tqdm.

True

Returns:

Type Description
list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Combined list of results, optionally with metadata.

Source code in pybdl/api/client.py
def fetch_all_results(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: bool = False,
    show_progress: bool = True,
) -> list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    """
    Fetch paginated results synchronously and combine them into a single list.

    Args:
        endpoint: API endpoint.
        method: HTTP method (default: GET).
        params: Query parameters.
        headers: Optional request headers.
        results_key: Key for extracting data from each page.
        page_size: Items per page.
        max_pages: Optional limit of pages.
        return_metadata: If True, return (results, metadata).
        show_progress: Display progress via tqdm.

    Returns:
        Combined list of results, optionally with metadata.
    """
    all_results: list[dict[str, Any]] = []
    metadata: dict[str, Any] = {}
    progress_bar = (
        tqdm(desc=f"Fetching {endpoint.split('/')[-1]}", unit=" pages", leave=True) if show_progress else None
    )

    first_page = True
    try:
        for page in self._paginated_request_sync(
            endpoint,
            method=method,
            params=params,
            headers=headers,
            results_key=results_key,
            page_size=page_size,
            max_pages=max_pages,
        ):
            if results_key not in page:
                raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=page)
            if first_page and return_metadata:
                metadata = self._metadata_from_response(page, results_key)
                if progress_bar is not None and "totalCount" in page:
                    total_pages = (page["totalCount"] + page_size - 1) // page_size
                    total_pages = min(total_pages, max_pages) if max_pages else total_pages
                    progress_bar.total = total_pages
                first_page = False

            all_results.extend(page.get(results_key, []))

            if progress_bar is not None:
                progress_bar.update(1)
                progress_bar.set_postfix({"items": len(all_results)})
    finally:
        if progress_bar is not None:
            progress_bar.close()

    return (all_results, metadata) if return_metadata else all_results

fetch_all_results_with_metadata

fetch_all_results_with_metadata(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    show_progress=True,
)
Source code in pybdl/api/client.py
def fetch_all_results_with_metadata(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    return self.fetch_all_results(
        endpoint,
        method=method,
        params=params,
        headers=headers,
        results_key=results_key,
        page_size=page_size,
        max_pages=max_pages,
        return_metadata=True,
        show_progress=show_progress,
    )

fetch_single_result

fetch_single_result(
    endpoint: str,
    *,
    results_key: None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> dict[str, Any]
fetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> list[dict[str, Any]]
fetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[True],
) -> tuple[list[dict[str, Any]], dict[str, Any]]
fetch_single_result(
    endpoint: str,
    *,
    results_key: None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[True],
) -> tuple[dict[str, Any], dict[str, Any]]
fetch_single_result(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
    return_metadata=False,
)

Fetch a single result, non-paginated (sync).

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
results_key str | None

If not None, extract this key from the JSON.

None
method str

HTTP method.

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
return_metadata bool

Also return metadata if True.

False

Returns:

Type Description
dict[str, Any] | list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]] | tuple[dict[str, Any], dict[str, Any]]

Dictionary or list, optionally with separate metadata.

Source code in pybdl/api/client.py
def fetch_single_result(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: bool = False,
) -> (
    dict[str, Any]
    | list[dict[str, Any]]
    | tuple[list[dict[str, Any]], dict[str, Any]]
    | tuple[dict[str, Any], dict[str, Any]]
):
    """
    Fetch a single result, non-paginated (sync).

    Args:
        endpoint: API endpoint.
        results_key: If not None, extract this key from the JSON.
        method: HTTP method.
        params: Query parameters.
        headers: Optional request headers.
        return_metadata: Also return metadata if True.

    Returns:
        Dictionary or list, optionally with separate metadata.
    """
    response = self._request_sync(
        endpoint=endpoint,
        method=method,
        params=params,
        headers=headers,
    )

    if results_key is None:
        if return_metadata:
            return response, {}
        return response

    if not isinstance(response, dict) or results_key not in response:
        raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=response)

    results_val = response[results_key]
    if return_metadata:
        metadata = self._metadata_from_response(response, results_key)
        return results_val, metadata

    return results_val

fetch_single_result_with_metadata

fetch_single_result_with_metadata(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
)
Source code in pybdl/api/client.py
def fetch_single_result_with_metadata(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    return self.fetch_single_result(
        endpoint,
        results_key=results_key,
        method=method,
        params=params,
        headers=headers,
        return_metadata=True,
    )

afetch_all_results async

afetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[False] = False,
    show_progress: bool = True,
) -> list[dict[str, Any]]
afetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[True],
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
afetch_all_results(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    return_metadata=False,
    show_progress=True,
)

Asynchronously fetch paginated results and combine them into a single list.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
method str

HTTP method (default: GET).

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
results_key str

Key for extracting data from each page.

'results'
page_size int

Items per page.

100
max_pages int | None

Optional limit of pages.

None
return_metadata bool

If True, return (results, metadata).

False
show_progress bool

Display progress via tqdm.

True

Returns:

Type Description
list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Combined list of results, optionally with metadata.

Source code in pybdl/api/client.py
async def afetch_all_results(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: bool = False,
    show_progress: bool = True,
) -> list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    """
    Asynchronously fetch paginated results and combine them into a single list.

    Args:
        endpoint: API endpoint.
        method: HTTP method (default: GET).
        params: Query parameters.
        headers: Optional request headers.
        results_key: Key for extracting data from each page.
        page_size: Items per page.
        max_pages: Optional limit of pages.
        return_metadata: If True, return (results, metadata).
        show_progress: Display progress via tqdm.

    Returns:
        Combined list of results, optionally with metadata.
    """
    all_results: list[dict[str, Any]] = []
    metadata: dict[str, Any] = {}
    first_page = True
    progress_bar = (
        tqdm(desc=f"Fetching {endpoint.split('/')[-1]} (async)", unit=" pages", leave=True)
        if show_progress
        else None
    )

    try:
        async for page in self._paginated_request_async(
            endpoint,
            method=method,
            params=params,
            headers=headers,
            results_key=results_key,
            page_size=page_size,
            max_pages=max_pages,
        ):
            if results_key not in page:
                raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=page)
            if first_page and return_metadata:
                metadata = self._metadata_from_response(page, results_key)
                if progress_bar is not None and "totalCount" in page:
                    total_pages = (page["totalCount"] + page_size - 1) // page_size
                    total_pages = min(total_pages, max_pages) if max_pages else total_pages
                    progress_bar.total = total_pages
                first_page = False

            all_results.extend(page.get(results_key, []))
            if progress_bar is not None:
                progress_bar.update(1)
                progress_bar.set_postfix({"items": len(all_results)})
    finally:
        if progress_bar is not None:
            progress_bar.close()

    return (all_results, metadata) if return_metadata else all_results

afetch_all_results_with_metadata async

afetch_all_results_with_metadata(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    show_progress=True,
)
Source code in pybdl/api/client.py
async def afetch_all_results_with_metadata(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    return await self.afetch_all_results(
        endpoint,
        method=method,
        params=params,
        headers=headers,
        results_key=results_key,
        page_size=page_size,
        max_pages=max_pages,
        return_metadata=True,
        show_progress=show_progress,
    )

afetch_single_result async

afetch_single_result(
    endpoint: str,
    *,
    method: str = "GET",
    results_key: Literal[None] = None,
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> dict[str, Any]
afetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> list[dict[str, Any]]
afetch_single_result(
    endpoint: str,
    *,
    return_metadata: Literal[True],
    results_key: Literal[None] = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]]
afetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    return_metadata: Literal[True],
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
afetch_single_result(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
    return_metadata=False,
)

Asynchronously fetch a single result, non-paginated.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
results_key str | None

If not None, extract this key from the JSON.

None
method str

HTTP method.

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
return_metadata bool

Also return metadata if True.

False

Returns:

Type Description
dict[str, Any] | list[dict[str, Any]] | tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Dictionary or list, optionally with separate metadata.

Source code in pybdl/api/client.py
async def afetch_single_result(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: bool = False,
) -> (
    dict[str, Any]
    | list[dict[str, Any]]
    | tuple[dict[str, Any], dict[str, Any]]
    | tuple[list[dict[str, Any]], dict[str, Any]]
):
    """
    Asynchronously fetch a single result, non-paginated.

    Args:
        endpoint: API endpoint.
        results_key: If not None, extract this key from the JSON.
        method: HTTP method.
        params: Query parameters.
        headers: Optional request headers.
        return_metadata: Also return metadata if True.

    Returns:
        Dictionary or list, optionally with separate metadata.
    """
    response = await self._request_async(
        endpoint=endpoint,
        method=method,
        params=params,
        headers=headers,
    )

    if results_key is None:
        return (response, {}) if return_metadata else response

    if not isinstance(response, dict) or results_key not in response:
        raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=response)

    results_val = cast(list[dict[str, Any]], response[results_key])
    if return_metadata:
        metadata = self._metadata_from_response(response, results_key)
        return results_val, metadata

    return results_val

afetch_single_result_with_metadata async

afetch_single_result_with_metadata(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
)
Source code in pybdl/api/client.py
async def afetch_single_result_with_metadata(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    return await self.afetch_single_result(
        endpoint,
        results_key=results_key,
        method=method,
        params=params,
        headers=headers,
        return_metadata=True,
    )

Subjects

subjects

SubjectsAPI

SubjectsAPI(config, extra_headers=None)

Bases: BaseAPIClient

Client for the BDL /subjects endpoints.

Initialize base API client for BDL.

Parameters:

Name Type Description Default
config BDLConfig

BDL configuration object.

required
extra_headers dict[str, str] | None

Optional extra headers (e.g., Accept-Language) to include in requests.

None
Source code in pybdl/api/client.py
def __init__(self, config: BDLConfig, extra_headers: dict[str, str] | None = None):
    """
    Initialize base API client for BDL.

    Args:
        config: BDL configuration object.
        extra_headers: Optional extra headers (e.g., Accept-Language) to include in requests.
    """
    self.config = config
    is_registered = bool(config.api_key)
    quotas: QuotaMap = cast(
        QuotaMap,
        config.custom_quotas if config.custom_quotas is not None else DEFAULT_QUOTAS,
    )
    self._quota_cache = PersistentQuotaCache(
        config.quota_cache_enabled,
        cache_file=config.quota_cache_file,
        use_global_cache=config.use_global_cache,
    )
    self._sync_limiter = RateLimiter(
        quotas,
        is_registered,
        self._quota_cache,
        raise_on_limit=config.raise_on_rate_limit,
    )
    self._async_limiter = AsyncRateLimiter(
        quotas,
        is_registered,
        self._quota_cache,
        raise_on_limit=config.raise_on_rate_limit,
    )
    self._proxy_url = self._build_proxy_url()
    self._http_cache_path = resolve_http_cache_db_path(config.cache_backend, self._quota_cache.cache_file)
    default_headers = self._build_default_headers(extra_headers)
    self.session = build_sync_http_client(
        cache_backend=config.cache_backend,
        http_cache_db_path=self._http_cache_path,
        default_headers=default_headers,
        proxy=self._proxy_url,
    )
    self._async_client = build_async_http_client(
        cache_backend=config.cache_backend,
        http_cache_db_path=self._http_cache_path,
        default_headers=default_headers,
        proxy=self._proxy_url,
    )

config instance-attribute

config = config

session instance-attribute

session = build_sync_http_client(
    cache_backend=cache_backend,
    http_cache_db_path=_http_cache_path,
    default_headers=default_headers,
    proxy=_proxy_url,
)

list_subjects

list_subjects(
    parent_id=None,
    sort=None,
    page=None,
    page_size=100,
    max_pages=None,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/subjects.py
def list_subjects(
    self,
    parent_id: str | None = None,
    sort: str | None = None,
    page: int | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    return self._fetch_collection_endpoint(
        "subjects",
        extra_params=self._list_params(parent_id, sort, page, extra_query),
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
        page_size=page_size,
        max_pages=max_pages,
        results_key="results",
    )

get_subject

get_subject(
    subject_id,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/subjects.py
def get_subject(
    self,
    subject_id: str,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return self._fetch_detail_endpoint(
        f"subjects/{subject_id}",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

search_subjects

search_subjects(
    name,
    page=None,
    page_size=100,
    max_pages=None,
    sort=None,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/subjects.py
def search_subjects(
    self,
    name: str,
    page: int | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    sort: str | None = None,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    return self._fetch_collection_endpoint(
        "subjects/search",
        extra_params=self._search_params(name, page, sort, extra_query),
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
        page_size=page_size,
        max_pages=max_pages,
        results_key="results",
    )

get_subjects_metadata

get_subjects_metadata(
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/subjects.py
def get_subjects_metadata(
    self,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return self._fetch_detail_endpoint(
        "subjects/metadata",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

alist_subjects async

alist_subjects(
    parent_id=None,
    sort=None,
    page=None,
    page_size=100,
    max_pages=None,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/subjects.py
async def alist_subjects(
    self,
    parent_id: str | None = None,
    sort: str | None = None,
    page: int | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    return await self._afetch_collection_endpoint(
        "subjects",
        extra_params=self._list_params(parent_id, sort, page, extra_query),
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
        page_size=page_size,
        max_pages=max_pages,
        results_key="results",
    )

aget_subject async

aget_subject(
    subject_id,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/subjects.py
async def aget_subject(
    self,
    subject_id: str,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return await self._afetch_detail_endpoint(
        f"subjects/{subject_id}",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

asearch_subjects async

asearch_subjects(
    name,
    page=None,
    page_size=100,
    max_pages=None,
    sort=None,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/subjects.py
async def asearch_subjects(
    self,
    name: str,
    page: int | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    sort: str | None = None,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    return await self._afetch_collection_endpoint(
        "subjects/search",
        extra_params=self._search_params(name, page, sort, extra_query),
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
        page_size=page_size,
        max_pages=max_pages,
        results_key="results",
    )

aget_subjects_metadata async

aget_subjects_metadata(
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/subjects.py
async def aget_subjects_metadata(
    self,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return await self._afetch_detail_endpoint(
        "subjects/metadata",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

close

close()

Close synchronous HTTP resources.

Source code in pybdl/api/client.py
def close(self) -> None:
    """Close synchronous HTTP resources."""
    self.session.close()

aclose async

aclose()

Close synchronous and asynchronous HTTP resources.

Source code in pybdl/api/client.py
async def aclose(self) -> None:
    """Close synchronous and asynchronous HTTP resources."""
    self.close()
    await self._async_client.aclose()

__enter__

__enter__()
Source code in pybdl/api/client.py
def __enter__(self) -> "BaseAPIClient":
    return self

__exit__

__exit__(exc_type, exc_val, exc_tb)
Source code in pybdl/api/client.py
def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> Literal[False]:
    self.close()
    return False

__aenter__ async

__aenter__()
Source code in pybdl/api/client.py
async def __aenter__(self) -> "BaseAPIClient":
    return self

__aexit__ async

__aexit__(exc_type, exc_val, exc_tb)
Source code in pybdl/api/client.py
async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> Literal[False]:
    await self.aclose()
    return False

fetch_all_results

fetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[False] = False,
    show_progress: bool = True,
) -> list[dict[str, Any]]
fetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[True],
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
fetch_all_results(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    return_metadata=False,
    show_progress=True,
)

Fetch paginated results synchronously and combine them into a single list.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
method str

HTTP method (default: GET).

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
results_key str

Key for extracting data from each page.

'results'
page_size int

Items per page.

100
max_pages int | None

Optional limit of pages.

None
return_metadata bool

If True, return (results, metadata).

False
show_progress bool

Display progress via tqdm.

True

Returns:

Type Description
list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Combined list of results, optionally with metadata.

Source code in pybdl/api/client.py
def fetch_all_results(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: bool = False,
    show_progress: bool = True,
) -> list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    """
    Fetch paginated results synchronously and combine them into a single list.

    Args:
        endpoint: API endpoint.
        method: HTTP method (default: GET).
        params: Query parameters.
        headers: Optional request headers.
        results_key: Key for extracting data from each page.
        page_size: Items per page.
        max_pages: Optional limit of pages.
        return_metadata: If True, return (results, metadata).
        show_progress: Display progress via tqdm.

    Returns:
        Combined list of results, optionally with metadata.
    """
    all_results: list[dict[str, Any]] = []
    metadata: dict[str, Any] = {}
    progress_bar = (
        tqdm(desc=f"Fetching {endpoint.split('/')[-1]}", unit=" pages", leave=True) if show_progress else None
    )

    first_page = True
    try:
        for page in self._paginated_request_sync(
            endpoint,
            method=method,
            params=params,
            headers=headers,
            results_key=results_key,
            page_size=page_size,
            max_pages=max_pages,
        ):
            if results_key not in page:
                raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=page)
            if first_page and return_metadata:
                metadata = self._metadata_from_response(page, results_key)
                if progress_bar is not None and "totalCount" in page:
                    total_pages = (page["totalCount"] + page_size - 1) // page_size
                    total_pages = min(total_pages, max_pages) if max_pages else total_pages
                    progress_bar.total = total_pages
                first_page = False

            all_results.extend(page.get(results_key, []))

            if progress_bar is not None:
                progress_bar.update(1)
                progress_bar.set_postfix({"items": len(all_results)})
    finally:
        if progress_bar is not None:
            progress_bar.close()

    return (all_results, metadata) if return_metadata else all_results

fetch_all_results_with_metadata

fetch_all_results_with_metadata(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    show_progress=True,
)
Source code in pybdl/api/client.py
def fetch_all_results_with_metadata(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    return self.fetch_all_results(
        endpoint,
        method=method,
        params=params,
        headers=headers,
        results_key=results_key,
        page_size=page_size,
        max_pages=max_pages,
        return_metadata=True,
        show_progress=show_progress,
    )

fetch_single_result

fetch_single_result(
    endpoint: str,
    *,
    results_key: None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> dict[str, Any]
fetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> list[dict[str, Any]]
fetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[True],
) -> tuple[list[dict[str, Any]], dict[str, Any]]
fetch_single_result(
    endpoint: str,
    *,
    results_key: None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[True],
) -> tuple[dict[str, Any], dict[str, Any]]
fetch_single_result(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
    return_metadata=False,
)

Fetch a single result, non-paginated (sync).

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
results_key str | None

If not None, extract this key from the JSON.

None
method str

HTTP method.

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
return_metadata bool

Also return metadata if True.

False

Returns:

Type Description
dict[str, Any] | list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]] | tuple[dict[str, Any], dict[str, Any]]

Dictionary or list, optionally with separate metadata.

Source code in pybdl/api/client.py
def fetch_single_result(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: bool = False,
) -> (
    dict[str, Any]
    | list[dict[str, Any]]
    | tuple[list[dict[str, Any]], dict[str, Any]]
    | tuple[dict[str, Any], dict[str, Any]]
):
    """
    Fetch a single result, non-paginated (sync).

    Args:
        endpoint: API endpoint.
        results_key: If not None, extract this key from the JSON.
        method: HTTP method.
        params: Query parameters.
        headers: Optional request headers.
        return_metadata: Also return metadata if True.

    Returns:
        Dictionary or list, optionally with separate metadata.
    """
    response = self._request_sync(
        endpoint=endpoint,
        method=method,
        params=params,
        headers=headers,
    )

    if results_key is None:
        if return_metadata:
            return response, {}
        return response

    if not isinstance(response, dict) or results_key not in response:
        raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=response)

    results_val = response[results_key]
    if return_metadata:
        metadata = self._metadata_from_response(response, results_key)
        return results_val, metadata

    return results_val

fetch_single_result_with_metadata

fetch_single_result_with_metadata(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
)
Source code in pybdl/api/client.py
def fetch_single_result_with_metadata(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    return self.fetch_single_result(
        endpoint,
        results_key=results_key,
        method=method,
        params=params,
        headers=headers,
        return_metadata=True,
    )

afetch_all_results async

afetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[False] = False,
    show_progress: bool = True,
) -> list[dict[str, Any]]
afetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[True],
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
afetch_all_results(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    return_metadata=False,
    show_progress=True,
)

Asynchronously fetch paginated results and combine them into a single list.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
method str

HTTP method (default: GET).

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
results_key str

Key for extracting data from each page.

'results'
page_size int

Items per page.

100
max_pages int | None

Optional limit of pages.

None
return_metadata bool

If True, return (results, metadata).

False
show_progress bool

Display progress via tqdm.

True

Returns:

Type Description
list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Combined list of results, optionally with metadata.

Source code in pybdl/api/client.py
async def afetch_all_results(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: bool = False,
    show_progress: bool = True,
) -> list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    """
    Asynchronously fetch paginated results and combine them into a single list.

    Args:
        endpoint: API endpoint.
        method: HTTP method (default: GET).
        params: Query parameters.
        headers: Optional request headers.
        results_key: Key for extracting data from each page.
        page_size: Items per page.
        max_pages: Optional limit of pages.
        return_metadata: If True, return (results, metadata).
        show_progress: Display progress via tqdm.

    Returns:
        Combined list of results, optionally with metadata.
    """
    all_results: list[dict[str, Any]] = []
    metadata: dict[str, Any] = {}
    first_page = True
    progress_bar = (
        tqdm(desc=f"Fetching {endpoint.split('/')[-1]} (async)", unit=" pages", leave=True)
        if show_progress
        else None
    )

    try:
        async for page in self._paginated_request_async(
            endpoint,
            method=method,
            params=params,
            headers=headers,
            results_key=results_key,
            page_size=page_size,
            max_pages=max_pages,
        ):
            if results_key not in page:
                raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=page)
            if first_page and return_metadata:
                metadata = self._metadata_from_response(page, results_key)
                if progress_bar is not None and "totalCount" in page:
                    total_pages = (page["totalCount"] + page_size - 1) // page_size
                    total_pages = min(total_pages, max_pages) if max_pages else total_pages
                    progress_bar.total = total_pages
                first_page = False

            all_results.extend(page.get(results_key, []))
            if progress_bar is not None:
                progress_bar.update(1)
                progress_bar.set_postfix({"items": len(all_results)})
    finally:
        if progress_bar is not None:
            progress_bar.close()

    return (all_results, metadata) if return_metadata else all_results

afetch_all_results_with_metadata async

afetch_all_results_with_metadata(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    show_progress=True,
)
Source code in pybdl/api/client.py
async def afetch_all_results_with_metadata(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    return await self.afetch_all_results(
        endpoint,
        method=method,
        params=params,
        headers=headers,
        results_key=results_key,
        page_size=page_size,
        max_pages=max_pages,
        return_metadata=True,
        show_progress=show_progress,
    )

afetch_single_result async

afetch_single_result(
    endpoint: str,
    *,
    method: str = "GET",
    results_key: Literal[None] = None,
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> dict[str, Any]
afetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> list[dict[str, Any]]
afetch_single_result(
    endpoint: str,
    *,
    return_metadata: Literal[True],
    results_key: Literal[None] = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]]
afetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    return_metadata: Literal[True],
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
afetch_single_result(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
    return_metadata=False,
)

Asynchronously fetch a single result, non-paginated.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
results_key str | None

If not None, extract this key from the JSON.

None
method str

HTTP method.

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
return_metadata bool

Also return metadata if True.

False

Returns:

Type Description
dict[str, Any] | list[dict[str, Any]] | tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Dictionary or list, optionally with separate metadata.

Source code in pybdl/api/client.py
async def afetch_single_result(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: bool = False,
) -> (
    dict[str, Any]
    | list[dict[str, Any]]
    | tuple[dict[str, Any], dict[str, Any]]
    | tuple[list[dict[str, Any]], dict[str, Any]]
):
    """
    Asynchronously fetch a single result, non-paginated.

    Args:
        endpoint: API endpoint.
        results_key: If not None, extract this key from the JSON.
        method: HTTP method.
        params: Query parameters.
        headers: Optional request headers.
        return_metadata: Also return metadata if True.

    Returns:
        Dictionary or list, optionally with separate metadata.
    """
    response = await self._request_async(
        endpoint=endpoint,
        method=method,
        params=params,
        headers=headers,
    )

    if results_key is None:
        return (response, {}) if return_metadata else response

    if not isinstance(response, dict) or results_key not in response:
        raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=response)

    results_val = cast(list[dict[str, Any]], response[results_key])
    if return_metadata:
        metadata = self._metadata_from_response(response, results_key)
        return results_val, metadata

    return results_val

afetch_single_result_with_metadata async

afetch_single_result_with_metadata(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
)
Source code in pybdl/api/client.py
async def afetch_single_result_with_metadata(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    return await self.afetch_single_result(
        endpoint,
        results_key=results_key,
        method=method,
        params=params,
        headers=headers,
        return_metadata=True,
    )

Units

units

UnitsAPI

UnitsAPI(config, extra_headers=None)

Bases: BaseAPIClient

Client for the BDL /units endpoints.

Initialize base API client for BDL.

Parameters:

Name Type Description Default
config BDLConfig

BDL configuration object.

required
extra_headers dict[str, str] | None

Optional extra headers (e.g., Accept-Language) to include in requests.

None
Source code in pybdl/api/client.py
def __init__(self, config: BDLConfig, extra_headers: dict[str, str] | None = None):
    """
    Initialize base API client for BDL.

    Args:
        config: BDL configuration object.
        extra_headers: Optional extra headers (e.g., Accept-Language) to include in requests.
    """
    self.config = config
    is_registered = bool(config.api_key)
    quotas: QuotaMap = cast(
        QuotaMap,
        config.custom_quotas if config.custom_quotas is not None else DEFAULT_QUOTAS,
    )
    self._quota_cache = PersistentQuotaCache(
        config.quota_cache_enabled,
        cache_file=config.quota_cache_file,
        use_global_cache=config.use_global_cache,
    )
    self._sync_limiter = RateLimiter(
        quotas,
        is_registered,
        self._quota_cache,
        raise_on_limit=config.raise_on_rate_limit,
    )
    self._async_limiter = AsyncRateLimiter(
        quotas,
        is_registered,
        self._quota_cache,
        raise_on_limit=config.raise_on_rate_limit,
    )
    self._proxy_url = self._build_proxy_url()
    self._http_cache_path = resolve_http_cache_db_path(config.cache_backend, self._quota_cache.cache_file)
    default_headers = self._build_default_headers(extra_headers)
    self.session = build_sync_http_client(
        cache_backend=config.cache_backend,
        http_cache_db_path=self._http_cache_path,
        default_headers=default_headers,
        proxy=self._proxy_url,
    )
    self._async_client = build_async_http_client(
        cache_backend=config.cache_backend,
        http_cache_db_path=self._http_cache_path,
        default_headers=default_headers,
        proxy=self._proxy_url,
    )

config instance-attribute

config = config

session instance-attribute

session = build_sync_http_client(
    cache_backend=cache_backend,
    http_cache_db_path=_http_cache_path,
    default_headers=default_headers,
    proxy=_proxy_url,
)

list_units

list_units(
    parent_id=None,
    level=None,
    page=None,
    page_size=100,
    max_pages=None,
    sort=None,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/units.py
def list_units(
    self,
    parent_id: str | None = None,
    level: list[int] | None = None,
    page: int | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    sort: str | None = None,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    return self._fetch_collection_endpoint(
        "units",
        extra_params=self._list_units_params(parent_id, level, page, sort, extra_query),
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
        page_size=page_size,
        max_pages=max_pages,
        results_key="results",
    )

get_unit

get_unit(
    unit_id,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/units.py
def get_unit(
    self,
    unit_id: str,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return self._fetch_detail_endpoint(
        f"units/{unit_id}",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

search_units

search_units(
    name=None,
    level=None,
    years=None,
    kind=None,
    page=None,
    page_size=100,
    max_pages=None,
    sort=None,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/units.py
def search_units(
    self,
    name: str | None = None,
    level: list[int] | None = None,
    years: list[int] | None = None,
    kind: str | None = None,
    page: int | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    sort: str | None = None,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    return self._fetch_collection_endpoint(
        "units/search",
        extra_params=self._search_units_params(name, level, years, kind, page, sort, extra_query),
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
        page_size=page_size,
        max_pages=max_pages,
        results_key="results",
    )

list_localities

list_localities(
    parent_id,
    page=None,
    page_size=100,
    max_pages=None,
    sort=None,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/units.py
def list_localities(
    self,
    parent_id: str,
    page: int | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    sort: str | None = None,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    return self._fetch_collection_endpoint(
        "units/localities",
        extra_params=self._list_localities_params(parent_id, page, sort, extra_query),
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
        page_size=page_size,
        max_pages=max_pages,
        results_key="results",
    )

get_locality

get_locality(
    locality_id,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/units.py
def get_locality(
    self,
    locality_id: str,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return self._fetch_detail_endpoint(
        f"units/localities/{locality_id}",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

search_localities

search_localities(
    name=None,
    years=None,
    page=None,
    page_size=100,
    max_pages=None,
    sort=None,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/units.py
def search_localities(
    self,
    name: str | None = None,
    years: list[int] | None = None,
    page: int | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    sort: str | None = None,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    return self._fetch_collection_endpoint(
        "units/localities/search",
        extra_params=self._search_localities_params(name, years, page, sort, extra_query),
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
        page_size=page_size,
        max_pages=max_pages,
        results_key="results",
    )

get_units_metadata

get_units_metadata(
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/units.py
def get_units_metadata(
    self,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return self._fetch_detail_endpoint(
        "units/metadata",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

alist_units async

alist_units(
    parent_id=None,
    level=None,
    page=None,
    page_size=100,
    max_pages=None,
    sort=None,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/units.py
async def alist_units(
    self,
    parent_id: str | None = None,
    level: list[int] | None = None,
    page: int | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    sort: str | None = None,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    return await self._afetch_collection_endpoint(
        "units",
        extra_params=self._list_units_params(parent_id, level, page, sort, extra_query),
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
        page_size=page_size,
        max_pages=max_pages,
        results_key="results",
    )

aget_unit async

aget_unit(
    unit_id,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/units.py
async def aget_unit(
    self,
    unit_id: str,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return await self._afetch_detail_endpoint(
        f"units/{unit_id}",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

asearch_units async

asearch_units(
    name=None,
    level=None,
    years=None,
    kind=None,
    page=None,
    page_size=100,
    max_pages=None,
    sort=None,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/units.py
async def asearch_units(
    self,
    name: str | None = None,
    level: list[int] | None = None,
    years: list[int] | None = None,
    kind: str | None = None,
    page: int | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    sort: str | None = None,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    return await self._afetch_collection_endpoint(
        "units/search",
        extra_params=self._search_units_params(name, level, years, kind, page, sort, extra_query),
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
        page_size=page_size,
        max_pages=max_pages,
        results_key="results",
    )

alist_localities async

alist_localities(
    parent_id,
    page=None,
    page_size=100,
    max_pages=None,
    sort=None,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/units.py
async def alist_localities(
    self,
    parent_id: str,
    page: int | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    sort: str | None = None,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    return await self._afetch_collection_endpoint(
        "units/localities",
        extra_params=self._list_localities_params(parent_id, page, sort, extra_query),
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
        page_size=page_size,
        max_pages=max_pages,
        results_key="results",
    )

aget_locality async

aget_locality(
    locality_id,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/units.py
async def aget_locality(
    self,
    locality_id: str,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return await self._afetch_detail_endpoint(
        f"units/localities/{locality_id}",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

asearch_localities async

asearch_localities(
    name=None,
    years=None,
    page=None,
    page_size=100,
    max_pages=None,
    sort=None,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/units.py
async def asearch_localities(
    self,
    name: str | None = None,
    years: list[int] | None = None,
    page: int | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    sort: str | None = None,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    return await self._afetch_collection_endpoint(
        "units/localities/search",
        extra_params=self._search_localities_params(name, years, page, sort, extra_query),
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
        page_size=page_size,
        max_pages=max_pages,
        results_key="results",
    )

aget_units_metadata async

aget_units_metadata(
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/units.py
async def aget_units_metadata(
    self,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return await self._afetch_detail_endpoint(
        "units/metadata",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

close

close()

Close synchronous HTTP resources.

Source code in pybdl/api/client.py
def close(self) -> None:
    """Close synchronous HTTP resources."""
    self.session.close()

aclose async

aclose()

Close synchronous and asynchronous HTTP resources.

Source code in pybdl/api/client.py
async def aclose(self) -> None:
    """Close synchronous and asynchronous HTTP resources."""
    self.close()
    await self._async_client.aclose()

__enter__

__enter__()
Source code in pybdl/api/client.py
def __enter__(self) -> "BaseAPIClient":
    return self

__exit__

__exit__(exc_type, exc_val, exc_tb)
Source code in pybdl/api/client.py
def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> Literal[False]:
    self.close()
    return False

__aenter__ async

__aenter__()
Source code in pybdl/api/client.py
async def __aenter__(self) -> "BaseAPIClient":
    return self

__aexit__ async

__aexit__(exc_type, exc_val, exc_tb)
Source code in pybdl/api/client.py
async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> Literal[False]:
    await self.aclose()
    return False

fetch_all_results

fetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[False] = False,
    show_progress: bool = True,
) -> list[dict[str, Any]]
fetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[True],
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
fetch_all_results(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    return_metadata=False,
    show_progress=True,
)

Fetch paginated results synchronously and combine them into a single list.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
method str

HTTP method (default: GET).

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
results_key str

Key for extracting data from each page.

'results'
page_size int

Items per page.

100
max_pages int | None

Optional limit of pages.

None
return_metadata bool

If True, return (results, metadata).

False
show_progress bool

Display progress via tqdm.

True

Returns:

Type Description
list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Combined list of results, optionally with metadata.

Source code in pybdl/api/client.py
def fetch_all_results(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: bool = False,
    show_progress: bool = True,
) -> list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    """
    Fetch paginated results synchronously and combine them into a single list.

    Args:
        endpoint: API endpoint.
        method: HTTP method (default: GET).
        params: Query parameters.
        headers: Optional request headers.
        results_key: Key for extracting data from each page.
        page_size: Items per page.
        max_pages: Optional limit of pages.
        return_metadata: If True, return (results, metadata).
        show_progress: Display progress via tqdm.

    Returns:
        Combined list of results, optionally with metadata.
    """
    all_results: list[dict[str, Any]] = []
    metadata: dict[str, Any] = {}
    progress_bar = (
        tqdm(desc=f"Fetching {endpoint.split('/')[-1]}", unit=" pages", leave=True) if show_progress else None
    )

    first_page = True
    try:
        for page in self._paginated_request_sync(
            endpoint,
            method=method,
            params=params,
            headers=headers,
            results_key=results_key,
            page_size=page_size,
            max_pages=max_pages,
        ):
            if results_key not in page:
                raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=page)
            if first_page and return_metadata:
                metadata = self._metadata_from_response(page, results_key)
                if progress_bar is not None and "totalCount" in page:
                    total_pages = (page["totalCount"] + page_size - 1) // page_size
                    total_pages = min(total_pages, max_pages) if max_pages else total_pages
                    progress_bar.total = total_pages
                first_page = False

            all_results.extend(page.get(results_key, []))

            if progress_bar is not None:
                progress_bar.update(1)
                progress_bar.set_postfix({"items": len(all_results)})
    finally:
        if progress_bar is not None:
            progress_bar.close()

    return (all_results, metadata) if return_metadata else all_results

fetch_all_results_with_metadata

fetch_all_results_with_metadata(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    show_progress=True,
)
Source code in pybdl/api/client.py
def fetch_all_results_with_metadata(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    return self.fetch_all_results(
        endpoint,
        method=method,
        params=params,
        headers=headers,
        results_key=results_key,
        page_size=page_size,
        max_pages=max_pages,
        return_metadata=True,
        show_progress=show_progress,
    )

fetch_single_result

fetch_single_result(
    endpoint: str,
    *,
    results_key: None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> dict[str, Any]
fetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> list[dict[str, Any]]
fetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[True],
) -> tuple[list[dict[str, Any]], dict[str, Any]]
fetch_single_result(
    endpoint: str,
    *,
    results_key: None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[True],
) -> tuple[dict[str, Any], dict[str, Any]]
fetch_single_result(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
    return_metadata=False,
)

Fetch a single result, non-paginated (sync).

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
results_key str | None

If not None, extract this key from the JSON.

None
method str

HTTP method.

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
return_metadata bool

Also return metadata if True.

False

Returns:

Type Description
dict[str, Any] | list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]] | tuple[dict[str, Any], dict[str, Any]]

Dictionary or list, optionally with separate metadata.

Source code in pybdl/api/client.py
def fetch_single_result(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: bool = False,
) -> (
    dict[str, Any]
    | list[dict[str, Any]]
    | tuple[list[dict[str, Any]], dict[str, Any]]
    | tuple[dict[str, Any], dict[str, Any]]
):
    """
    Fetch a single result, non-paginated (sync).

    Args:
        endpoint: API endpoint.
        results_key: If not None, extract this key from the JSON.
        method: HTTP method.
        params: Query parameters.
        headers: Optional request headers.
        return_metadata: Also return metadata if True.

    Returns:
        Dictionary or list, optionally with separate metadata.
    """
    response = self._request_sync(
        endpoint=endpoint,
        method=method,
        params=params,
        headers=headers,
    )

    if results_key is None:
        if return_metadata:
            return response, {}
        return response

    if not isinstance(response, dict) or results_key not in response:
        raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=response)

    results_val = response[results_key]
    if return_metadata:
        metadata = self._metadata_from_response(response, results_key)
        return results_val, metadata

    return results_val

fetch_single_result_with_metadata

fetch_single_result_with_metadata(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
)
Source code in pybdl/api/client.py
def fetch_single_result_with_metadata(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    return self.fetch_single_result(
        endpoint,
        results_key=results_key,
        method=method,
        params=params,
        headers=headers,
        return_metadata=True,
    )

afetch_all_results async

afetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[False] = False,
    show_progress: bool = True,
) -> list[dict[str, Any]]
afetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[True],
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
afetch_all_results(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    return_metadata=False,
    show_progress=True,
)

Asynchronously fetch paginated results and combine them into a single list.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
method str

HTTP method (default: GET).

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
results_key str

Key for extracting data from each page.

'results'
page_size int

Items per page.

100
max_pages int | None

Optional limit of pages.

None
return_metadata bool

If True, return (results, metadata).

False
show_progress bool

Display progress via tqdm.

True

Returns:

Type Description
list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Combined list of results, optionally with metadata.

Source code in pybdl/api/client.py
async def afetch_all_results(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: bool = False,
    show_progress: bool = True,
) -> list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    """
    Asynchronously fetch paginated results and combine them into a single list.

    Args:
        endpoint: API endpoint.
        method: HTTP method (default: GET).
        params: Query parameters.
        headers: Optional request headers.
        results_key: Key for extracting data from each page.
        page_size: Items per page.
        max_pages: Optional limit of pages.
        return_metadata: If True, return (results, metadata).
        show_progress: Display progress via tqdm.

    Returns:
        Combined list of results, optionally with metadata.
    """
    all_results: list[dict[str, Any]] = []
    metadata: dict[str, Any] = {}
    first_page = True
    progress_bar = (
        tqdm(desc=f"Fetching {endpoint.split('/')[-1]} (async)", unit=" pages", leave=True)
        if show_progress
        else None
    )

    try:
        async for page in self._paginated_request_async(
            endpoint,
            method=method,
            params=params,
            headers=headers,
            results_key=results_key,
            page_size=page_size,
            max_pages=max_pages,
        ):
            if results_key not in page:
                raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=page)
            if first_page and return_metadata:
                metadata = self._metadata_from_response(page, results_key)
                if progress_bar is not None and "totalCount" in page:
                    total_pages = (page["totalCount"] + page_size - 1) // page_size
                    total_pages = min(total_pages, max_pages) if max_pages else total_pages
                    progress_bar.total = total_pages
                first_page = False

            all_results.extend(page.get(results_key, []))
            if progress_bar is not None:
                progress_bar.update(1)
                progress_bar.set_postfix({"items": len(all_results)})
    finally:
        if progress_bar is not None:
            progress_bar.close()

    return (all_results, metadata) if return_metadata else all_results

afetch_all_results_with_metadata async

afetch_all_results_with_metadata(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    show_progress=True,
)
Source code in pybdl/api/client.py
async def afetch_all_results_with_metadata(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    return await self.afetch_all_results(
        endpoint,
        method=method,
        params=params,
        headers=headers,
        results_key=results_key,
        page_size=page_size,
        max_pages=max_pages,
        return_metadata=True,
        show_progress=show_progress,
    )

afetch_single_result async

afetch_single_result(
    endpoint: str,
    *,
    method: str = "GET",
    results_key: Literal[None] = None,
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> dict[str, Any]
afetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> list[dict[str, Any]]
afetch_single_result(
    endpoint: str,
    *,
    return_metadata: Literal[True],
    results_key: Literal[None] = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]]
afetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    return_metadata: Literal[True],
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
afetch_single_result(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
    return_metadata=False,
)

Asynchronously fetch a single result, non-paginated.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
results_key str | None

If not None, extract this key from the JSON.

None
method str

HTTP method.

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
return_metadata bool

Also return metadata if True.

False

Returns:

Type Description
dict[str, Any] | list[dict[str, Any]] | tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Dictionary or list, optionally with separate metadata.

Source code in pybdl/api/client.py
async def afetch_single_result(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: bool = False,
) -> (
    dict[str, Any]
    | list[dict[str, Any]]
    | tuple[dict[str, Any], dict[str, Any]]
    | tuple[list[dict[str, Any]], dict[str, Any]]
):
    """
    Asynchronously fetch a single result, non-paginated.

    Args:
        endpoint: API endpoint.
        results_key: If not None, extract this key from the JSON.
        method: HTTP method.
        params: Query parameters.
        headers: Optional request headers.
        return_metadata: Also return metadata if True.

    Returns:
        Dictionary or list, optionally with separate metadata.
    """
    response = await self._request_async(
        endpoint=endpoint,
        method=method,
        params=params,
        headers=headers,
    )

    if results_key is None:
        return (response, {}) if return_metadata else response

    if not isinstance(response, dict) or results_key not in response:
        raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=response)

    results_val = cast(list[dict[str, Any]], response[results_key])
    if return_metadata:
        metadata = self._metadata_from_response(response, results_key)
        return results_val, metadata

    return results_val

afetch_single_result_with_metadata async

afetch_single_result_with_metadata(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
)
Source code in pybdl/api/client.py
async def afetch_single_result_with_metadata(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    return await self.afetch_single_result(
        endpoint,
        results_key=results_key,
        method=method,
        params=params,
        headers=headers,
        return_metadata=True,
    )

Variables

variables

VariablesAPI

VariablesAPI(config, extra_headers=None)

Bases: BaseAPIClient

Client for the BDL /variables endpoints.

Initialize base API client for BDL.

Parameters:

Name Type Description Default
config BDLConfig

BDL configuration object.

required
extra_headers dict[str, str] | None

Optional extra headers (e.g., Accept-Language) to include in requests.

None
Source code in pybdl/api/client.py
def __init__(self, config: BDLConfig, extra_headers: dict[str, str] | None = None):
    """
    Initialize base API client for BDL.

    Args:
        config: BDL configuration object.
        extra_headers: Optional extra headers (e.g., Accept-Language) to include in requests.
    """
    self.config = config
    is_registered = bool(config.api_key)
    quotas: QuotaMap = cast(
        QuotaMap,
        config.custom_quotas if config.custom_quotas is not None else DEFAULT_QUOTAS,
    )
    self._quota_cache = PersistentQuotaCache(
        config.quota_cache_enabled,
        cache_file=config.quota_cache_file,
        use_global_cache=config.use_global_cache,
    )
    self._sync_limiter = RateLimiter(
        quotas,
        is_registered,
        self._quota_cache,
        raise_on_limit=config.raise_on_rate_limit,
    )
    self._async_limiter = AsyncRateLimiter(
        quotas,
        is_registered,
        self._quota_cache,
        raise_on_limit=config.raise_on_rate_limit,
    )
    self._proxy_url = self._build_proxy_url()
    self._http_cache_path = resolve_http_cache_db_path(config.cache_backend, self._quota_cache.cache_file)
    default_headers = self._build_default_headers(extra_headers)
    self.session = build_sync_http_client(
        cache_backend=config.cache_backend,
        http_cache_db_path=self._http_cache_path,
        default_headers=default_headers,
        proxy=self._proxy_url,
    )
    self._async_client = build_async_http_client(
        cache_backend=config.cache_backend,
        http_cache_db_path=self._http_cache_path,
        default_headers=default_headers,
        proxy=self._proxy_url,
    )

config instance-attribute

config = config

session instance-attribute

session = build_sync_http_client(
    cache_backend=cache_backend,
    http_cache_db_path=_http_cache_path,
    default_headers=default_headers,
    proxy=_proxy_url,
)

list_variables

list_variables(
    subject_id=None,
    level=None,
    years=None,
    page=None,
    page_size=100,
    max_pages=None,
    sort=None,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/variables.py
def list_variables(
    self,
    subject_id: str | None = None,
    level: int | None = None,
    years: list[int] | None = None,
    page: int | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    sort: str | None = None,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    return self._fetch_collection_endpoint(
        "variables",
        extra_params=self._list_params(subject_id, level, years, page, sort, extra_query),
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
        page_size=page_size,
        max_pages=max_pages,
        results_key="results",
    )

get_variable

get_variable(
    variable_id,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/variables.py
def get_variable(
    self,
    variable_id: str,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return self._fetch_detail_endpoint(
        f"variables/{variable_id}",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

search_variables

search_variables(
    name=None,
    subject_id=None,
    level=None,
    years=None,
    page=None,
    page_size=100,
    max_pages=None,
    sort=None,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/variables.py
def search_variables(
    self,
    name: str | None = None,
    subject_id: str | None = None,
    level: int | None = None,
    years: list[int] | None = None,
    page: int | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    sort: str | None = None,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    return self._fetch_collection_endpoint(
        "variables/search",
        extra_params=self._search_params(name, subject_id, level, years, page, sort, extra_query),
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
        page_size=page_size,
        max_pages=max_pages,
        results_key="results",
    )

get_variables_metadata

get_variables_metadata(
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/variables.py
def get_variables_metadata(
    self,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return self._fetch_detail_endpoint(
        "variables/metadata",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

alist_variables async

alist_variables(
    subject_id=None,
    level=None,
    years=None,
    page=None,
    page_size=100,
    max_pages=None,
    sort=None,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/variables.py
async def alist_variables(
    self,
    subject_id: str | None = None,
    level: int | None = None,
    years: list[int] | None = None,
    page: int | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    sort: str | None = None,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    return await self._afetch_collection_endpoint(
        "variables",
        extra_params=self._list_params(subject_id, level, years, page, sort, extra_query),
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
        page_size=page_size,
        max_pages=max_pages,
        results_key="results",
    )

aget_variable async

aget_variable(
    variable_id,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/variables.py
async def aget_variable(
    self,
    variable_id: str,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return await self._afetch_detail_endpoint(
        f"variables/{variable_id}",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

asearch_variables async

asearch_variables(
    name=None,
    subject_id=None,
    level=None,
    years=None,
    page=None,
    page_size=100,
    max_pages=None,
    sort=None,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/variables.py
async def asearch_variables(
    self,
    name: str | None = None,
    subject_id: str | None = None,
    level: int | None = None,
    years: list[int] | None = None,
    page: int | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    sort: str | None = None,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    return await self._afetch_collection_endpoint(
        "variables/search",
        extra_params=self._search_params(name, subject_id, level, years, page, sort, extra_query),
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
        page_size=page_size,
        max_pages=max_pages,
        results_key="results",
    )

aget_variables_metadata async

aget_variables_metadata(
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/variables.py
async def aget_variables_metadata(
    self,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return await self._afetch_detail_endpoint(
        "variables/metadata",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

close

close()

Close synchronous HTTP resources.

Source code in pybdl/api/client.py
def close(self) -> None:
    """Close synchronous HTTP resources."""
    self.session.close()

aclose async

aclose()

Close synchronous and asynchronous HTTP resources.

Source code in pybdl/api/client.py
async def aclose(self) -> None:
    """Close synchronous and asynchronous HTTP resources."""
    self.close()
    await self._async_client.aclose()

__enter__

__enter__()
Source code in pybdl/api/client.py
def __enter__(self) -> "BaseAPIClient":
    return self

__exit__

__exit__(exc_type, exc_val, exc_tb)
Source code in pybdl/api/client.py
def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> Literal[False]:
    self.close()
    return False

__aenter__ async

__aenter__()
Source code in pybdl/api/client.py
async def __aenter__(self) -> "BaseAPIClient":
    return self

__aexit__ async

__aexit__(exc_type, exc_val, exc_tb)
Source code in pybdl/api/client.py
async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> Literal[False]:
    await self.aclose()
    return False

fetch_all_results

fetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[False] = False,
    show_progress: bool = True,
) -> list[dict[str, Any]]
fetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[True],
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
fetch_all_results(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    return_metadata=False,
    show_progress=True,
)

Fetch paginated results synchronously and combine them into a single list.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
method str

HTTP method (default: GET).

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
results_key str

Key for extracting data from each page.

'results'
page_size int

Items per page.

100
max_pages int | None

Optional limit of pages.

None
return_metadata bool

If True, return (results, metadata).

False
show_progress bool

Display progress via tqdm.

True

Returns:

Type Description
list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Combined list of results, optionally with metadata.

Source code in pybdl/api/client.py
def fetch_all_results(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: bool = False,
    show_progress: bool = True,
) -> list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    """
    Fetch paginated results synchronously and combine them into a single list.

    Args:
        endpoint: API endpoint.
        method: HTTP method (default: GET).
        params: Query parameters.
        headers: Optional request headers.
        results_key: Key for extracting data from each page.
        page_size: Items per page.
        max_pages: Optional limit of pages.
        return_metadata: If True, return (results, metadata).
        show_progress: Display progress via tqdm.

    Returns:
        Combined list of results, optionally with metadata.
    """
    all_results: list[dict[str, Any]] = []
    metadata: dict[str, Any] = {}
    progress_bar = (
        tqdm(desc=f"Fetching {endpoint.split('/')[-1]}", unit=" pages", leave=True) if show_progress else None
    )

    first_page = True
    try:
        for page in self._paginated_request_sync(
            endpoint,
            method=method,
            params=params,
            headers=headers,
            results_key=results_key,
            page_size=page_size,
            max_pages=max_pages,
        ):
            if results_key not in page:
                raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=page)
            if first_page and return_metadata:
                metadata = self._metadata_from_response(page, results_key)
                if progress_bar is not None and "totalCount" in page:
                    total_pages = (page["totalCount"] + page_size - 1) // page_size
                    total_pages = min(total_pages, max_pages) if max_pages else total_pages
                    progress_bar.total = total_pages
                first_page = False

            all_results.extend(page.get(results_key, []))

            if progress_bar is not None:
                progress_bar.update(1)
                progress_bar.set_postfix({"items": len(all_results)})
    finally:
        if progress_bar is not None:
            progress_bar.close()

    return (all_results, metadata) if return_metadata else all_results

fetch_all_results_with_metadata

fetch_all_results_with_metadata(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    show_progress=True,
)
Source code in pybdl/api/client.py
def fetch_all_results_with_metadata(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    return self.fetch_all_results(
        endpoint,
        method=method,
        params=params,
        headers=headers,
        results_key=results_key,
        page_size=page_size,
        max_pages=max_pages,
        return_metadata=True,
        show_progress=show_progress,
    )

fetch_single_result

fetch_single_result(
    endpoint: str,
    *,
    results_key: None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> dict[str, Any]
fetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> list[dict[str, Any]]
fetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[True],
) -> tuple[list[dict[str, Any]], dict[str, Any]]
fetch_single_result(
    endpoint: str,
    *,
    results_key: None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[True],
) -> tuple[dict[str, Any], dict[str, Any]]
fetch_single_result(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
    return_metadata=False,
)

Fetch a single result, non-paginated (sync).

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
results_key str | None

If not None, extract this key from the JSON.

None
method str

HTTP method.

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
return_metadata bool

Also return metadata if True.

False

Returns:

Type Description
dict[str, Any] | list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]] | tuple[dict[str, Any], dict[str, Any]]

Dictionary or list, optionally with separate metadata.

Source code in pybdl/api/client.py
def fetch_single_result(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: bool = False,
) -> (
    dict[str, Any]
    | list[dict[str, Any]]
    | tuple[list[dict[str, Any]], dict[str, Any]]
    | tuple[dict[str, Any], dict[str, Any]]
):
    """
    Fetch a single result, non-paginated (sync).

    Args:
        endpoint: API endpoint.
        results_key: If not None, extract this key from the JSON.
        method: HTTP method.
        params: Query parameters.
        headers: Optional request headers.
        return_metadata: Also return metadata if True.

    Returns:
        Dictionary or list, optionally with separate metadata.
    """
    response = self._request_sync(
        endpoint=endpoint,
        method=method,
        params=params,
        headers=headers,
    )

    if results_key is None:
        if return_metadata:
            return response, {}
        return response

    if not isinstance(response, dict) or results_key not in response:
        raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=response)

    results_val = response[results_key]
    if return_metadata:
        metadata = self._metadata_from_response(response, results_key)
        return results_val, metadata

    return results_val

fetch_single_result_with_metadata

fetch_single_result_with_metadata(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
)
Source code in pybdl/api/client.py
def fetch_single_result_with_metadata(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    return self.fetch_single_result(
        endpoint,
        results_key=results_key,
        method=method,
        params=params,
        headers=headers,
        return_metadata=True,
    )

afetch_all_results async

afetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[False] = False,
    show_progress: bool = True,
) -> list[dict[str, Any]]
afetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[True],
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
afetch_all_results(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    return_metadata=False,
    show_progress=True,
)

Asynchronously fetch paginated results and combine them into a single list.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
method str

HTTP method (default: GET).

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
results_key str

Key for extracting data from each page.

'results'
page_size int

Items per page.

100
max_pages int | None

Optional limit of pages.

None
return_metadata bool

If True, return (results, metadata).

False
show_progress bool

Display progress via tqdm.

True

Returns:

Type Description
list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Combined list of results, optionally with metadata.

Source code in pybdl/api/client.py
async def afetch_all_results(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: bool = False,
    show_progress: bool = True,
) -> list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    """
    Asynchronously fetch paginated results and combine them into a single list.

    Args:
        endpoint: API endpoint.
        method: HTTP method (default: GET).
        params: Query parameters.
        headers: Optional request headers.
        results_key: Key for extracting data from each page.
        page_size: Items per page.
        max_pages: Optional limit of pages.
        return_metadata: If True, return (results, metadata).
        show_progress: Display progress via tqdm.

    Returns:
        Combined list of results, optionally with metadata.
    """
    all_results: list[dict[str, Any]] = []
    metadata: dict[str, Any] = {}
    first_page = True
    progress_bar = (
        tqdm(desc=f"Fetching {endpoint.split('/')[-1]} (async)", unit=" pages", leave=True)
        if show_progress
        else None
    )

    try:
        async for page in self._paginated_request_async(
            endpoint,
            method=method,
            params=params,
            headers=headers,
            results_key=results_key,
            page_size=page_size,
            max_pages=max_pages,
        ):
            if results_key not in page:
                raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=page)
            if first_page and return_metadata:
                metadata = self._metadata_from_response(page, results_key)
                if progress_bar is not None and "totalCount" in page:
                    total_pages = (page["totalCount"] + page_size - 1) // page_size
                    total_pages = min(total_pages, max_pages) if max_pages else total_pages
                    progress_bar.total = total_pages
                first_page = False

            all_results.extend(page.get(results_key, []))
            if progress_bar is not None:
                progress_bar.update(1)
                progress_bar.set_postfix({"items": len(all_results)})
    finally:
        if progress_bar is not None:
            progress_bar.close()

    return (all_results, metadata) if return_metadata else all_results

afetch_all_results_with_metadata async

afetch_all_results_with_metadata(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    show_progress=True,
)
Source code in pybdl/api/client.py
async def afetch_all_results_with_metadata(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    return await self.afetch_all_results(
        endpoint,
        method=method,
        params=params,
        headers=headers,
        results_key=results_key,
        page_size=page_size,
        max_pages=max_pages,
        return_metadata=True,
        show_progress=show_progress,
    )

afetch_single_result async

afetch_single_result(
    endpoint: str,
    *,
    method: str = "GET",
    results_key: Literal[None] = None,
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> dict[str, Any]
afetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> list[dict[str, Any]]
afetch_single_result(
    endpoint: str,
    *,
    return_metadata: Literal[True],
    results_key: Literal[None] = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]]
afetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    return_metadata: Literal[True],
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
afetch_single_result(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
    return_metadata=False,
)

Asynchronously fetch a single result, non-paginated.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
results_key str | None

If not None, extract this key from the JSON.

None
method str

HTTP method.

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
return_metadata bool

Also return metadata if True.

False

Returns:

Type Description
dict[str, Any] | list[dict[str, Any]] | tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Dictionary or list, optionally with separate metadata.

Source code in pybdl/api/client.py
async def afetch_single_result(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: bool = False,
) -> (
    dict[str, Any]
    | list[dict[str, Any]]
    | tuple[dict[str, Any], dict[str, Any]]
    | tuple[list[dict[str, Any]], dict[str, Any]]
):
    """
    Asynchronously fetch a single result, non-paginated.

    Args:
        endpoint: API endpoint.
        results_key: If not None, extract this key from the JSON.
        method: HTTP method.
        params: Query parameters.
        headers: Optional request headers.
        return_metadata: Also return metadata if True.

    Returns:
        Dictionary or list, optionally with separate metadata.
    """
    response = await self._request_async(
        endpoint=endpoint,
        method=method,
        params=params,
        headers=headers,
    )

    if results_key is None:
        return (response, {}) if return_metadata else response

    if not isinstance(response, dict) or results_key not in response:
        raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=response)

    results_val = cast(list[dict[str, Any]], response[results_key])
    if return_metadata:
        metadata = self._metadata_from_response(response, results_key)
        return results_val, metadata

    return results_val

afetch_single_result_with_metadata async

afetch_single_result_with_metadata(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
)
Source code in pybdl/api/client.py
async def afetch_single_result_with_metadata(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    return await self.afetch_single_result(
        endpoint,
        results_key=results_key,
        method=method,
        params=params,
        headers=headers,
        return_metadata=True,
    )

Version

version

VersionAPI

VersionAPI(config, extra_headers=None)

Bases: BaseAPIClient

Client for the BDL /version endpoint.

Initialize base API client for BDL.

Parameters:

Name Type Description Default
config BDLConfig

BDL configuration object.

required
extra_headers dict[str, str] | None

Optional extra headers (e.g., Accept-Language) to include in requests.

None
Source code in pybdl/api/client.py
def __init__(self, config: BDLConfig, extra_headers: dict[str, str] | None = None):
    """
    Initialize base API client for BDL.

    Args:
        config: BDL configuration object.
        extra_headers: Optional extra headers (e.g., Accept-Language) to include in requests.
    """
    self.config = config
    is_registered = bool(config.api_key)
    quotas: QuotaMap = cast(
        QuotaMap,
        config.custom_quotas if config.custom_quotas is not None else DEFAULT_QUOTAS,
    )
    self._quota_cache = PersistentQuotaCache(
        config.quota_cache_enabled,
        cache_file=config.quota_cache_file,
        use_global_cache=config.use_global_cache,
    )
    self._sync_limiter = RateLimiter(
        quotas,
        is_registered,
        self._quota_cache,
        raise_on_limit=config.raise_on_rate_limit,
    )
    self._async_limiter = AsyncRateLimiter(
        quotas,
        is_registered,
        self._quota_cache,
        raise_on_limit=config.raise_on_rate_limit,
    )
    self._proxy_url = self._build_proxy_url()
    self._http_cache_path = resolve_http_cache_db_path(config.cache_backend, self._quota_cache.cache_file)
    default_headers = self._build_default_headers(extra_headers)
    self.session = build_sync_http_client(
        cache_backend=config.cache_backend,
        http_cache_db_path=self._http_cache_path,
        default_headers=default_headers,
        proxy=self._proxy_url,
    )
    self._async_client = build_async_http_client(
        cache_backend=config.cache_backend,
        http_cache_db_path=self._http_cache_path,
        default_headers=default_headers,
        proxy=self._proxy_url,
    )

config instance-attribute

config = config

session instance-attribute

session = build_sync_http_client(
    cache_backend=cache_backend,
    http_cache_db_path=_http_cache_path,
    default_headers=default_headers,
    proxy=_proxy_url,
)

get_version

get_version(
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/version.py
def get_version(
    self,
    lang: LanguageLiteral | None = None,
    format: Literal["json", "xml"] | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return self._fetch_detail_endpoint(
        "version",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

aget_version async

aget_version(
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/version.py
async def aget_version(
    self,
    lang: LanguageLiteral | None = None,
    format: Literal["json", "xml"] | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return await self._afetch_detail_endpoint(
        "version",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

close

close()

Close synchronous HTTP resources.

Source code in pybdl/api/client.py
def close(self) -> None:
    """Close synchronous HTTP resources."""
    self.session.close()

aclose async

aclose()

Close synchronous and asynchronous HTTP resources.

Source code in pybdl/api/client.py
async def aclose(self) -> None:
    """Close synchronous and asynchronous HTTP resources."""
    self.close()
    await self._async_client.aclose()

__enter__

__enter__()
Source code in pybdl/api/client.py
def __enter__(self) -> "BaseAPIClient":
    return self

__exit__

__exit__(exc_type, exc_val, exc_tb)
Source code in pybdl/api/client.py
def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> Literal[False]:
    self.close()
    return False

__aenter__ async

__aenter__()
Source code in pybdl/api/client.py
async def __aenter__(self) -> "BaseAPIClient":
    return self

__aexit__ async

__aexit__(exc_type, exc_val, exc_tb)
Source code in pybdl/api/client.py
async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> Literal[False]:
    await self.aclose()
    return False

fetch_all_results

fetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[False] = False,
    show_progress: bool = True,
) -> list[dict[str, Any]]
fetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[True],
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
fetch_all_results(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    return_metadata=False,
    show_progress=True,
)

Fetch paginated results synchronously and combine them into a single list.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
method str

HTTP method (default: GET).

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
results_key str

Key for extracting data from each page.

'results'
page_size int

Items per page.

100
max_pages int | None

Optional limit of pages.

None
return_metadata bool

If True, return (results, metadata).

False
show_progress bool

Display progress via tqdm.

True

Returns:

Type Description
list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Combined list of results, optionally with metadata.

Source code in pybdl/api/client.py
def fetch_all_results(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: bool = False,
    show_progress: bool = True,
) -> list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    """
    Fetch paginated results synchronously and combine them into a single list.

    Args:
        endpoint: API endpoint.
        method: HTTP method (default: GET).
        params: Query parameters.
        headers: Optional request headers.
        results_key: Key for extracting data from each page.
        page_size: Items per page.
        max_pages: Optional limit of pages.
        return_metadata: If True, return (results, metadata).
        show_progress: Display progress via tqdm.

    Returns:
        Combined list of results, optionally with metadata.
    """
    all_results: list[dict[str, Any]] = []
    metadata: dict[str, Any] = {}
    progress_bar = (
        tqdm(desc=f"Fetching {endpoint.split('/')[-1]}", unit=" pages", leave=True) if show_progress else None
    )

    first_page = True
    try:
        for page in self._paginated_request_sync(
            endpoint,
            method=method,
            params=params,
            headers=headers,
            results_key=results_key,
            page_size=page_size,
            max_pages=max_pages,
        ):
            if results_key not in page:
                raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=page)
            if first_page and return_metadata:
                metadata = self._metadata_from_response(page, results_key)
                if progress_bar is not None and "totalCount" in page:
                    total_pages = (page["totalCount"] + page_size - 1) // page_size
                    total_pages = min(total_pages, max_pages) if max_pages else total_pages
                    progress_bar.total = total_pages
                first_page = False

            all_results.extend(page.get(results_key, []))

            if progress_bar is not None:
                progress_bar.update(1)
                progress_bar.set_postfix({"items": len(all_results)})
    finally:
        if progress_bar is not None:
            progress_bar.close()

    return (all_results, metadata) if return_metadata else all_results

fetch_all_results_with_metadata

fetch_all_results_with_metadata(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    show_progress=True,
)
Source code in pybdl/api/client.py
def fetch_all_results_with_metadata(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    return self.fetch_all_results(
        endpoint,
        method=method,
        params=params,
        headers=headers,
        results_key=results_key,
        page_size=page_size,
        max_pages=max_pages,
        return_metadata=True,
        show_progress=show_progress,
    )

fetch_single_result

fetch_single_result(
    endpoint: str,
    *,
    results_key: None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> dict[str, Any]
fetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> list[dict[str, Any]]
fetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[True],
) -> tuple[list[dict[str, Any]], dict[str, Any]]
fetch_single_result(
    endpoint: str,
    *,
    results_key: None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[True],
) -> tuple[dict[str, Any], dict[str, Any]]
fetch_single_result(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
    return_metadata=False,
)

Fetch a single result, non-paginated (sync).

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
results_key str | None

If not None, extract this key from the JSON.

None
method str

HTTP method.

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
return_metadata bool

Also return metadata if True.

False

Returns:

Type Description
dict[str, Any] | list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]] | tuple[dict[str, Any], dict[str, Any]]

Dictionary or list, optionally with separate metadata.

Source code in pybdl/api/client.py
def fetch_single_result(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: bool = False,
) -> (
    dict[str, Any]
    | list[dict[str, Any]]
    | tuple[list[dict[str, Any]], dict[str, Any]]
    | tuple[dict[str, Any], dict[str, Any]]
):
    """
    Fetch a single result, non-paginated (sync).

    Args:
        endpoint: API endpoint.
        results_key: If not None, extract this key from the JSON.
        method: HTTP method.
        params: Query parameters.
        headers: Optional request headers.
        return_metadata: Also return metadata if True.

    Returns:
        Dictionary or list, optionally with separate metadata.
    """
    response = self._request_sync(
        endpoint=endpoint,
        method=method,
        params=params,
        headers=headers,
    )

    if results_key is None:
        if return_metadata:
            return response, {}
        return response

    if not isinstance(response, dict) or results_key not in response:
        raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=response)

    results_val = response[results_key]
    if return_metadata:
        metadata = self._metadata_from_response(response, results_key)
        return results_val, metadata

    return results_val

fetch_single_result_with_metadata

fetch_single_result_with_metadata(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
)
Source code in pybdl/api/client.py
def fetch_single_result_with_metadata(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    return self.fetch_single_result(
        endpoint,
        results_key=results_key,
        method=method,
        params=params,
        headers=headers,
        return_metadata=True,
    )

afetch_all_results async

afetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[False] = False,
    show_progress: bool = True,
) -> list[dict[str, Any]]
afetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[True],
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
afetch_all_results(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    return_metadata=False,
    show_progress=True,
)

Asynchronously fetch paginated results and combine them into a single list.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
method str

HTTP method (default: GET).

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
results_key str

Key for extracting data from each page.

'results'
page_size int

Items per page.

100
max_pages int | None

Optional limit of pages.

None
return_metadata bool

If True, return (results, metadata).

False
show_progress bool

Display progress via tqdm.

True

Returns:

Type Description
list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Combined list of results, optionally with metadata.

Source code in pybdl/api/client.py
async def afetch_all_results(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: bool = False,
    show_progress: bool = True,
) -> list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    """
    Asynchronously fetch paginated results and combine them into a single list.

    Args:
        endpoint: API endpoint.
        method: HTTP method (default: GET).
        params: Query parameters.
        headers: Optional request headers.
        results_key: Key for extracting data from each page.
        page_size: Items per page.
        max_pages: Optional limit of pages.
        return_metadata: If True, return (results, metadata).
        show_progress: Display progress via tqdm.

    Returns:
        Combined list of results, optionally with metadata.
    """
    all_results: list[dict[str, Any]] = []
    metadata: dict[str, Any] = {}
    first_page = True
    progress_bar = (
        tqdm(desc=f"Fetching {endpoint.split('/')[-1]} (async)", unit=" pages", leave=True)
        if show_progress
        else None
    )

    try:
        async for page in self._paginated_request_async(
            endpoint,
            method=method,
            params=params,
            headers=headers,
            results_key=results_key,
            page_size=page_size,
            max_pages=max_pages,
        ):
            if results_key not in page:
                raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=page)
            if first_page and return_metadata:
                metadata = self._metadata_from_response(page, results_key)
                if progress_bar is not None and "totalCount" in page:
                    total_pages = (page["totalCount"] + page_size - 1) // page_size
                    total_pages = min(total_pages, max_pages) if max_pages else total_pages
                    progress_bar.total = total_pages
                first_page = False

            all_results.extend(page.get(results_key, []))
            if progress_bar is not None:
                progress_bar.update(1)
                progress_bar.set_postfix({"items": len(all_results)})
    finally:
        if progress_bar is not None:
            progress_bar.close()

    return (all_results, metadata) if return_metadata else all_results

afetch_all_results_with_metadata async

afetch_all_results_with_metadata(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    show_progress=True,
)
Source code in pybdl/api/client.py
async def afetch_all_results_with_metadata(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    return await self.afetch_all_results(
        endpoint,
        method=method,
        params=params,
        headers=headers,
        results_key=results_key,
        page_size=page_size,
        max_pages=max_pages,
        return_metadata=True,
        show_progress=show_progress,
    )

afetch_single_result async

afetch_single_result(
    endpoint: str,
    *,
    method: str = "GET",
    results_key: Literal[None] = None,
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> dict[str, Any]
afetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> list[dict[str, Any]]
afetch_single_result(
    endpoint: str,
    *,
    return_metadata: Literal[True],
    results_key: Literal[None] = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]]
afetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    return_metadata: Literal[True],
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
afetch_single_result(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
    return_metadata=False,
)

Asynchronously fetch a single result, non-paginated.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
results_key str | None

If not None, extract this key from the JSON.

None
method str

HTTP method.

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
return_metadata bool

Also return metadata if True.

False

Returns:

Type Description
dict[str, Any] | list[dict[str, Any]] | tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Dictionary or list, optionally with separate metadata.

Source code in pybdl/api/client.py
async def afetch_single_result(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: bool = False,
) -> (
    dict[str, Any]
    | list[dict[str, Any]]
    | tuple[dict[str, Any], dict[str, Any]]
    | tuple[list[dict[str, Any]], dict[str, Any]]
):
    """
    Asynchronously fetch a single result, non-paginated.

    Args:
        endpoint: API endpoint.
        results_key: If not None, extract this key from the JSON.
        method: HTTP method.
        params: Query parameters.
        headers: Optional request headers.
        return_metadata: Also return metadata if True.

    Returns:
        Dictionary or list, optionally with separate metadata.
    """
    response = await self._request_async(
        endpoint=endpoint,
        method=method,
        params=params,
        headers=headers,
    )

    if results_key is None:
        return (response, {}) if return_metadata else response

    if not isinstance(response, dict) or results_key not in response:
        raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=response)

    results_val = cast(list[dict[str, Any]], response[results_key])
    if return_metadata:
        metadata = self._metadata_from_response(response, results_key)
        return results_val, metadata

    return results_val

afetch_single_result_with_metadata async

afetch_single_result_with_metadata(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
)
Source code in pybdl/api/client.py
async def afetch_single_result_with_metadata(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    return await self.afetch_single_result(
        endpoint,
        results_key=results_key,
        method=method,
        params=params,
        headers=headers,
        return_metadata=True,
    )

Years

years

YearsAPI

YearsAPI(config, extra_headers=None)

Bases: BaseAPIClient

Client for the BDL /years endpoints.

Initialize base API client for BDL.

Parameters:

Name Type Description Default
config BDLConfig

BDL configuration object.

required
extra_headers dict[str, str] | None

Optional extra headers (e.g., Accept-Language) to include in requests.

None
Source code in pybdl/api/client.py
def __init__(self, config: BDLConfig, extra_headers: dict[str, str] | None = None):
    """
    Initialize base API client for BDL.

    Args:
        config: BDL configuration object.
        extra_headers: Optional extra headers (e.g., Accept-Language) to include in requests.
    """
    self.config = config
    is_registered = bool(config.api_key)
    quotas: QuotaMap = cast(
        QuotaMap,
        config.custom_quotas if config.custom_quotas is not None else DEFAULT_QUOTAS,
    )
    self._quota_cache = PersistentQuotaCache(
        config.quota_cache_enabled,
        cache_file=config.quota_cache_file,
        use_global_cache=config.use_global_cache,
    )
    self._sync_limiter = RateLimiter(
        quotas,
        is_registered,
        self._quota_cache,
        raise_on_limit=config.raise_on_rate_limit,
    )
    self._async_limiter = AsyncRateLimiter(
        quotas,
        is_registered,
        self._quota_cache,
        raise_on_limit=config.raise_on_rate_limit,
    )
    self._proxy_url = self._build_proxy_url()
    self._http_cache_path = resolve_http_cache_db_path(config.cache_backend, self._quota_cache.cache_file)
    default_headers = self._build_default_headers(extra_headers)
    self.session = build_sync_http_client(
        cache_backend=config.cache_backend,
        http_cache_db_path=self._http_cache_path,
        default_headers=default_headers,
        proxy=self._proxy_url,
    )
    self._async_client = build_async_http_client(
        cache_backend=config.cache_backend,
        http_cache_db_path=self._http_cache_path,
        default_headers=default_headers,
        proxy=self._proxy_url,
    )

config instance-attribute

config = config

session instance-attribute

session = build_sync_http_client(
    cache_backend=cache_backend,
    http_cache_db_path=_http_cache_path,
    default_headers=default_headers,
    proxy=_proxy_url,
)

list_years

list_years(
    sort=None,
    page_size=100,
    max_pages=None,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/years.py
def list_years(
    self,
    sort: str | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    return self._fetch_collection_endpoint(
        "years",
        extra_params=self._list_params(sort, extra_query),
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
        page_size=page_size,
        max_pages=max_pages,
    )

get_year

get_year(
    year_id,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/years.py
def get_year(
    self,
    year_id: int,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return self._fetch_detail_endpoint(
        f"years/{year_id}",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

get_years_metadata

get_years_metadata(
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/years.py
def get_years_metadata(
    self,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return self._fetch_detail_endpoint(
        "years/metadata",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

alist_years async

alist_years(
    sort=None,
    page_size=100,
    max_pages=None,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/years.py
async def alist_years(
    self,
    sort: str | None = None,
    page_size: int = 100,
    max_pages: int | None = None,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    return await self._afetch_collection_endpoint(
        "years",
        extra_params=self._list_params(sort, extra_query),
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
        page_size=page_size,
        max_pages=max_pages,
    )

aget_year async

aget_year(
    year_id,
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/years.py
async def aget_year(
    self,
    year_id: int,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return await self._afetch_detail_endpoint(
        f"years/{year_id}",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

aget_years_metadata async

aget_years_metadata(
    lang=None,
    format=None,
    if_none_match=None,
    if_modified_since=None,
    extra_query=None,
)
Source code in pybdl/api/years.py
async def aget_years_metadata(
    self,
    lang: LanguageLiteral | None = None,
    format: FormatLiteral | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
    extra_query: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return await self._afetch_detail_endpoint(
        "years/metadata",
        extra_params=extra_query,
        lang=lang,
        format=format,
        if_none_match=if_none_match,
        if_modified_since=if_modified_since,
    )

close

close()

Close synchronous HTTP resources.

Source code in pybdl/api/client.py
def close(self) -> None:
    """Close synchronous HTTP resources."""
    self.session.close()

aclose async

aclose()

Close synchronous and asynchronous HTTP resources.

Source code in pybdl/api/client.py
async def aclose(self) -> None:
    """Close synchronous and asynchronous HTTP resources."""
    self.close()
    await self._async_client.aclose()

__enter__

__enter__()
Source code in pybdl/api/client.py
def __enter__(self) -> "BaseAPIClient":
    return self

__exit__

__exit__(exc_type, exc_val, exc_tb)
Source code in pybdl/api/client.py
def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> Literal[False]:
    self.close()
    return False

__aenter__ async

__aenter__()
Source code in pybdl/api/client.py
async def __aenter__(self) -> "BaseAPIClient":
    return self

__aexit__ async

__aexit__(exc_type, exc_val, exc_tb)
Source code in pybdl/api/client.py
async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> Literal[False]:
    await self.aclose()
    return False

fetch_all_results

fetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[False] = False,
    show_progress: bool = True,
) -> list[dict[str, Any]]
fetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[True],
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
fetch_all_results(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    return_metadata=False,
    show_progress=True,
)

Fetch paginated results synchronously and combine them into a single list.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
method str

HTTP method (default: GET).

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
results_key str

Key for extracting data from each page.

'results'
page_size int

Items per page.

100
max_pages int | None

Optional limit of pages.

None
return_metadata bool

If True, return (results, metadata).

False
show_progress bool

Display progress via tqdm.

True

Returns:

Type Description
list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Combined list of results, optionally with metadata.

Source code in pybdl/api/client.py
def fetch_all_results(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: bool = False,
    show_progress: bool = True,
) -> list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    """
    Fetch paginated results synchronously and combine them into a single list.

    Args:
        endpoint: API endpoint.
        method: HTTP method (default: GET).
        params: Query parameters.
        headers: Optional request headers.
        results_key: Key for extracting data from each page.
        page_size: Items per page.
        max_pages: Optional limit of pages.
        return_metadata: If True, return (results, metadata).
        show_progress: Display progress via tqdm.

    Returns:
        Combined list of results, optionally with metadata.
    """
    all_results: list[dict[str, Any]] = []
    metadata: dict[str, Any] = {}
    progress_bar = (
        tqdm(desc=f"Fetching {endpoint.split('/')[-1]}", unit=" pages", leave=True) if show_progress else None
    )

    first_page = True
    try:
        for page in self._paginated_request_sync(
            endpoint,
            method=method,
            params=params,
            headers=headers,
            results_key=results_key,
            page_size=page_size,
            max_pages=max_pages,
        ):
            if results_key not in page:
                raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=page)
            if first_page and return_metadata:
                metadata = self._metadata_from_response(page, results_key)
                if progress_bar is not None and "totalCount" in page:
                    total_pages = (page["totalCount"] + page_size - 1) // page_size
                    total_pages = min(total_pages, max_pages) if max_pages else total_pages
                    progress_bar.total = total_pages
                first_page = False

            all_results.extend(page.get(results_key, []))

            if progress_bar is not None:
                progress_bar.update(1)
                progress_bar.set_postfix({"items": len(all_results)})
    finally:
        if progress_bar is not None:
            progress_bar.close()

    return (all_results, metadata) if return_metadata else all_results

fetch_all_results_with_metadata

fetch_all_results_with_metadata(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    show_progress=True,
)
Source code in pybdl/api/client.py
def fetch_all_results_with_metadata(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    return self.fetch_all_results(
        endpoint,
        method=method,
        params=params,
        headers=headers,
        results_key=results_key,
        page_size=page_size,
        max_pages=max_pages,
        return_metadata=True,
        show_progress=show_progress,
    )

fetch_single_result

fetch_single_result(
    endpoint: str,
    *,
    results_key: None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> dict[str, Any]
fetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> list[dict[str, Any]]
fetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[True],
) -> tuple[list[dict[str, Any]], dict[str, Any]]
fetch_single_result(
    endpoint: str,
    *,
    results_key: None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[True],
) -> tuple[dict[str, Any], dict[str, Any]]
fetch_single_result(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
    return_metadata=False,
)

Fetch a single result, non-paginated (sync).

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
results_key str | None

If not None, extract this key from the JSON.

None
method str

HTTP method.

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
return_metadata bool

Also return metadata if True.

False

Returns:

Type Description
dict[str, Any] | list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]] | tuple[dict[str, Any], dict[str, Any]]

Dictionary or list, optionally with separate metadata.

Source code in pybdl/api/client.py
def fetch_single_result(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: bool = False,
) -> (
    dict[str, Any]
    | list[dict[str, Any]]
    | tuple[list[dict[str, Any]], dict[str, Any]]
    | tuple[dict[str, Any], dict[str, Any]]
):
    """
    Fetch a single result, non-paginated (sync).

    Args:
        endpoint: API endpoint.
        results_key: If not None, extract this key from the JSON.
        method: HTTP method.
        params: Query parameters.
        headers: Optional request headers.
        return_metadata: Also return metadata if True.

    Returns:
        Dictionary or list, optionally with separate metadata.
    """
    response = self._request_sync(
        endpoint=endpoint,
        method=method,
        params=params,
        headers=headers,
    )

    if results_key is None:
        if return_metadata:
            return response, {}
        return response

    if not isinstance(response, dict) or results_key not in response:
        raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=response)

    results_val = response[results_key]
    if return_metadata:
        metadata = self._metadata_from_response(response, results_key)
        return results_val, metadata

    return results_val

fetch_single_result_with_metadata

fetch_single_result_with_metadata(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
)
Source code in pybdl/api/client.py
def fetch_single_result_with_metadata(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    return self.fetch_single_result(
        endpoint,
        results_key=results_key,
        method=method,
        params=params,
        headers=headers,
        return_metadata=True,
    )

afetch_all_results async

afetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[False] = False,
    show_progress: bool = True,
) -> list[dict[str, Any]]
afetch_all_results(
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: Literal[True],
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
afetch_all_results(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    return_metadata=False,
    show_progress=True,
)

Asynchronously fetch paginated results and combine them into a single list.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
method str

HTTP method (default: GET).

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
results_key str

Key for extracting data from each page.

'results'
page_size int

Items per page.

100
max_pages int | None

Optional limit of pages.

None
return_metadata bool

If True, return (results, metadata).

False
show_progress bool

Display progress via tqdm.

True

Returns:

Type Description
list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Combined list of results, optionally with metadata.

Source code in pybdl/api/client.py
async def afetch_all_results(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    return_metadata: bool = False,
    show_progress: bool = True,
) -> list[dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    """
    Asynchronously fetch paginated results and combine them into a single list.

    Args:
        endpoint: API endpoint.
        method: HTTP method (default: GET).
        params: Query parameters.
        headers: Optional request headers.
        results_key: Key for extracting data from each page.
        page_size: Items per page.
        max_pages: Optional limit of pages.
        return_metadata: If True, return (results, metadata).
        show_progress: Display progress via tqdm.

    Returns:
        Combined list of results, optionally with metadata.
    """
    all_results: list[dict[str, Any]] = []
    metadata: dict[str, Any] = {}
    first_page = True
    progress_bar = (
        tqdm(desc=f"Fetching {endpoint.split('/')[-1]} (async)", unit=" pages", leave=True)
        if show_progress
        else None
    )

    try:
        async for page in self._paginated_request_async(
            endpoint,
            method=method,
            params=params,
            headers=headers,
            results_key=results_key,
            page_size=page_size,
            max_pages=max_pages,
        ):
            if results_key not in page:
                raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=page)
            if first_page and return_metadata:
                metadata = self._metadata_from_response(page, results_key)
                if progress_bar is not None and "totalCount" in page:
                    total_pages = (page["totalCount"] + page_size - 1) // page_size
                    total_pages = min(total_pages, max_pages) if max_pages else total_pages
                    progress_bar.total = total_pages
                first_page = False

            all_results.extend(page.get(results_key, []))
            if progress_bar is not None:
                progress_bar.update(1)
                progress_bar.set_postfix({"items": len(all_results)})
    finally:
        if progress_bar is not None:
            progress_bar.close()

    return (all_results, metadata) if return_metadata else all_results

afetch_all_results_with_metadata async

afetch_all_results_with_metadata(
    endpoint,
    *,
    method="GET",
    params=None,
    headers=None,
    results_key="results",
    page_size=100,
    max_pages=None,
    show_progress=True,
)
Source code in pybdl/api/client.py
async def afetch_all_results_with_metadata(
    self,
    endpoint: str,
    *,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    results_key: str = "results",
    page_size: int = 100,
    max_pages: int | None = None,
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    return await self.afetch_all_results(
        endpoint,
        method=method,
        params=params,
        headers=headers,
        results_key=results_key,
        page_size=page_size,
        max_pages=max_pages,
        return_metadata=True,
        show_progress=show_progress,
    )

afetch_single_result async

afetch_single_result(
    endpoint: str,
    *,
    method: str = "GET",
    results_key: Literal[None] = None,
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> dict[str, Any]
afetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: Literal[False] = False,
) -> list[dict[str, Any]]
afetch_single_result(
    endpoint: str,
    *,
    return_metadata: Literal[True],
    results_key: Literal[None] = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]]
afetch_single_result(
    endpoint: str,
    *,
    results_key: str,
    return_metadata: Literal[True],
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[list[dict[str, Any]], dict[str, Any]]
afetch_single_result(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
    return_metadata=False,
)

Asynchronously fetch a single result, non-paginated.

Parameters:

Name Type Description Default
endpoint str

API endpoint.

required
results_key str | None

If not None, extract this key from the JSON.

None
method str

HTTP method.

'GET'
params dict[str, Any] | None

Query parameters.

None
headers dict[str, str] | None

Optional request headers.

None
return_metadata bool

Also return metadata if True.

False

Returns:

Type Description
dict[str, Any] | list[dict[str, Any]] | tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]

Dictionary or list, optionally with separate metadata.

Source code in pybdl/api/client.py
async def afetch_single_result(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    return_metadata: bool = False,
) -> (
    dict[str, Any]
    | list[dict[str, Any]]
    | tuple[dict[str, Any], dict[str, Any]]
    | tuple[list[dict[str, Any]], dict[str, Any]]
):
    """
    Asynchronously fetch a single result, non-paginated.

    Args:
        endpoint: API endpoint.
        results_key: If not None, extract this key from the JSON.
        method: HTTP method.
        params: Query parameters.
        headers: Optional request headers.
        return_metadata: Also return metadata if True.

    Returns:
        Dictionary or list, optionally with separate metadata.
    """
    response = await self._request_async(
        endpoint=endpoint,
        method=method,
        params=params,
        headers=headers,
    )

    if results_key is None:
        return (response, {}) if return_metadata else response

    if not isinstance(response, dict) or results_key not in response:
        raise BDLResponseError(f"Response does not contain key '{results_key}'", payload=response)

    results_val = cast(list[dict[str, Any]], response[results_key])
    if return_metadata:
        metadata = self._metadata_from_response(response, results_key)
        return results_val, metadata

    return results_val

afetch_single_result_with_metadata async

afetch_single_result_with_metadata(
    endpoint,
    *,
    results_key=None,
    method="GET",
    params=None,
    headers=None,
)
Source code in pybdl/api/client.py
async def afetch_single_result_with_metadata(
    self,
    endpoint: str,
    *,
    results_key: str | None = None,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]] | tuple[list[dict[str, Any]], dict[str, Any]]:
    return await self.afetch_single_result(
        endpoint,
        results_key=results_key,
        method=method,
        params=params,
        headers=headers,
        return_metadata=True,
    )