Source code for pybdl.api.measures

from typing import Any

from pybdl.api.client import (
    BaseAPIClient,
    FormatLiteral,
    LanguageLiteral,
)


[docs] class MeasuresAPI(BaseAPIClient): """ Client for the BDL /measures endpoints. Provides access to measure unit metadata (e.g. "number", "percent", "kg") used for variables in the Local Data Bank (BDL). """
[docs] 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]]: """ List all measure units, optionally sorted. Maps to: GET /measures Args: sort: Optional sorting order, e.g. 'Id', '-Id', 'Name', '-Name'. page_size: Number of results per page. max_pages: Maximum number of pages to fetch (None for all). lang: Expected response content language (defaults to config.language). format: Expected response content type (defaults to config.format). if_none_match: Conditional request header If-None-Match (entity tag). if_modified_since: Conditional request header If-Modified-Since. extra_query: Additional query parameters. Returns: List of measure unit metadata dictionaries. """ extra_params: dict[str, Any] = {} if sort: extra_params["sort"] = sort if extra_query: extra_params.update(extra_query) params, headers = self._prepare_api_params_and_headers( lang=lang, format=format, if_none_match=if_none_match, if_modified_since=if_modified_since, extra_params=extra_params, ) return self.fetch_all_results( "measures", params=params, headers=headers if headers else None, page_size=page_size, max_pages=max_pages )
[docs] 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]: """ Retrieve metadata for a specific measure unit. Maps to: GET /measures/{id} Args: measure_id: Measure unit identifier (integer). lang: Expected response content language (defaults to config.language). format: Expected response content type (defaults to config.format). if_none_match: Conditional request header If-None-Match (entity tag). if_modified_since: Conditional request header If-Modified-Since. extra_query: Additional query parameters. Returns: Dictionary with measure unit metadata. """ params, headers = self._prepare_api_params_and_headers( lang=lang, format=format, if_none_match=if_none_match, if_modified_since=if_modified_since, extra_params=extra_query, ) return self.fetch_single_result( f"measures/{measure_id}", params=params if params else None, headers=headers if headers else None )
[docs] 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]: """ Retrieve general metadata and version information for the /measures endpoint. Maps to: GET /measures/metadata Args: lang: Expected response content language (defaults to config.language). format: Expected response content type (defaults to config.format). if_none_match: Conditional request header If-None-Match (entity tag). if_modified_since: Conditional request header If-Modified-Since. extra_query: Additional query parameters. Returns: Dictionary with endpoint metadata and versioning info. """ params, headers = self._prepare_api_params_and_headers( lang=lang, format=format, if_none_match=if_none_match, if_modified_since=if_modified_since, extra_params=extra_query, ) return self.fetch_single_result( "measures/metadata", params=params if params else None, headers=headers if headers else None )
[docs] 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]]: """ Asynchronously list all measure units, optionally sorted. Maps to: GET /measures Args: sort: Optional sorting order, e.g. 'Id', '-Id', 'Name', '-Name'. page_size: Number of results per page. max_pages: Maximum number of pages to fetch (None for all). lang: Expected response content language (defaults to config.language). format: Expected response content type (defaults to config.format). if_none_match: Conditional request header If-None-Match (entity tag). if_modified_since: Conditional request header If-Modified-Since. extra_query: Additional query parameters. Returns: List of measure unit metadata dictionaries. """ extra_params: dict[str, Any] = {} if sort: extra_params["sort"] = sort if extra_query: extra_params.update(extra_query) params, headers = self._prepare_api_params_and_headers( lang=lang, format=format, if_none_match=if_none_match, if_modified_since=if_modified_since, extra_params=extra_params, ) return await self.afetch_all_results( "measures", params=params, headers=headers if headers else None, page_size=page_size, max_pages=max_pages )
[docs] 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]: """ Asynchronously retrieve metadata for a specific measure unit. Maps to: GET /measures/{id} Args: measure_id: Measure unit identifier (integer). lang: Expected response content language (defaults to config.language). format: Expected response content type (defaults to config.format). if_none_match: Conditional request header If-None-Match (entity tag). if_modified_since: Conditional request header If-Modified-Since. extra_query: Additional query parameters. Returns: Dictionary with measure unit metadata. """ params, headers = self._prepare_api_params_and_headers( lang=lang, format=format, if_none_match=if_none_match, if_modified_since=if_modified_since, extra_params=extra_query, ) return await self.afetch_single_result( f"measures/{measure_id}", params=params if params else None, headers=headers if headers else None )
[docs] 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]: """ Asynchronously retrieve general metadata and version information for the /measures endpoint. Maps to: GET /measures/metadata Args: lang: Expected response content language (defaults to config.language). format: Expected response content type (defaults to config.format). if_none_match: Conditional request header If-None-Match (entity tag). if_modified_since: Conditional request header If-Modified-Since. extra_query: Additional query parameters. Returns: Dictionary with endpoint metadata and versioning info. """ params, headers = self._prepare_api_params_and_headers( lang=lang, format=format, if_none_match=if_none_match, if_modified_since=if_modified_since, extra_params=extra_query, ) return await self.afetch_single_result( "measures/metadata", params=params if params else None, headers=headers if headers else None )