Skip to content

API Reference

Clients

esiosapy.ESIOSAPYClient

A client for interacting with the ESIOS API.

This client provides access to various managers that handle specific types of requests to the ESIOS API, such as archives, indicators, and offer indicators. It simplifies the process of making requests by managing authentication and constructing the necessary URLs.

Can be used as a context manager:

with ESIOSAPYClient(token="xxx") as client:
    indicators = client.indicators.list_all()
Source code in esiosapy/client.py
class ESIOSAPYClient:
    """
    A client for interacting with the ESIOS API.

    This client provides access to various managers that handle specific
    types of requests to the ESIOS API, such as archives, indicators, and
    offer indicators. It simplifies the process of making requests by
    managing authentication and constructing the necessary URLs.

    Can be used as a context manager:

        with ESIOSAPYClient(token="xxx") as client:
            indicators = client.indicators.list_all()
    """

    def __init__(self, token: str, base_url: str = ESIOS_API_URL, timeout: int = 30):
        """
        Initializes the ESIOSAPYClient with an API token and a base URL.

        :param token: The API token used for authentication.
        :type token: str
        :param base_url: The base URL for the ESIOS API. Defaults to ESIOS_API_URL.
        :type base_url: str, optional
        :param timeout: Request timeout in seconds, defaults to 30.
        :type timeout: int
        """
        self.token = token
        self.base_url = base_url
        self.timeout = timeout
        self.request_helper = RequestHelper(base_url, token, timeout)

        self.archives: ArchiveManager = ArchiveManager(self.request_helper)
        self.indicators: IndicatorManager = IndicatorManager(self.request_helper)
        self.offer_indicators: OfferIndicatorManager = OfferIndicatorManager(
            self.request_helper
        )

    def close(self) -> None:
        """Close the client and release resources."""
        self.request_helper._session.close()

    def __enter__(self) -> ESIOSAPYClient:
        """Enter the context manager."""
        return self

    def __exit__(self, *args: Any) -> None:
        """Exit the context manager and close the session."""
        self.close()

    def raw_request(
        self, url: str, headers: dict[str, str] | None = None
    ) -> requests.Response:
        """
        Makes a raw GET request to a specified URL with optional headers.

        This method allows for making a direct GET request to a specified URL.
        It adds default headers to the request and handles URL construction
        if the provided URL is relative.

        :param url: The URL to which the GET request is made. If the URL is
                    relative, it will be joined with the base URL.
        :type url: str
        :param headers: Optional headers to include in the request. If not provided,
                        default headers will be added. Defaults to None.
        :type headers: Optional[Dict[str, str]], optional
        :return: The response object resulting from the GET request.
        :rtype: requests.Response
        """
        if headers is None:
            headers = {}
        headers = self.request_helper.add_default_headers(headers)

        if urlparse(url).netloc == "":
            url = urljoin(self.base_url, url)

        return requests.get(url, headers=headers, timeout=self.timeout)

Functions

__enter__()

Enter the context manager.

Source code in esiosapy/client.py
def __enter__(self) -> ESIOSAPYClient:
    """Enter the context manager."""
    return self

__exit__(*args)

Exit the context manager and close the session.

Source code in esiosapy/client.py
def __exit__(self, *args: Any) -> None:
    """Exit the context manager and close the session."""
    self.close()

__init__(token, base_url=ESIOS_API_URL, timeout=30)

Initializes the ESIOSAPYClient with an API token and a base URL.

Parameters:

Name Type Description Default
token str

The API token used for authentication.

required
base_url str

The base URL for the ESIOS API. Defaults to ESIOS_API_URL.

ESIOS_API_URL
timeout int

Request timeout in seconds, defaults to 30.

30
Source code in esiosapy/client.py
def __init__(self, token: str, base_url: str = ESIOS_API_URL, timeout: int = 30):
    """
    Initializes the ESIOSAPYClient with an API token and a base URL.

    :param token: The API token used for authentication.
    :type token: str
    :param base_url: The base URL for the ESIOS API. Defaults to ESIOS_API_URL.
    :type base_url: str, optional
    :param timeout: Request timeout in seconds, defaults to 30.
    :type timeout: int
    """
    self.token = token
    self.base_url = base_url
    self.timeout = timeout
    self.request_helper = RequestHelper(base_url, token, timeout)

    self.archives: ArchiveManager = ArchiveManager(self.request_helper)
    self.indicators: IndicatorManager = IndicatorManager(self.request_helper)
    self.offer_indicators: OfferIndicatorManager = OfferIndicatorManager(
        self.request_helper
    )

close()

Close the client and release resources.

Source code in esiosapy/client.py
def close(self) -> None:
    """Close the client and release resources."""
    self.request_helper._session.close()

raw_request(url, headers=None)

Makes a raw GET request to a specified URL with optional headers.

This method allows for making a direct GET request to a specified URL. It adds default headers to the request and handles URL construction if the provided URL is relative.

Parameters:

Name Type Description Default
url str

The URL to which the GET request is made. If the URL is relative, it will be joined with the base URL.

required
headers dict[str, str] | None

Optional headers to include in the request. If not provided, default headers will be added. Defaults to None.

None

Returns:

Type Description
requests.Response

The response object resulting from the GET request.

Source code in esiosapy/client.py
def raw_request(
    self, url: str, headers: dict[str, str] | None = None
) -> requests.Response:
    """
    Makes a raw GET request to a specified URL with optional headers.

    This method allows for making a direct GET request to a specified URL.
    It adds default headers to the request and handles URL construction
    if the provided URL is relative.

    :param url: The URL to which the GET request is made. If the URL is
                relative, it will be joined with the base URL.
    :type url: str
    :param headers: Optional headers to include in the request. If not provided,
                    default headers will be added. Defaults to None.
    :type headers: Optional[Dict[str, str]], optional
    :return: The response object resulting from the GET request.
    :rtype: requests.Response
    """
    if headers is None:
        headers = {}
    headers = self.request_helper.add_default_headers(headers)

    if urlparse(url).netloc == "":
        url = urljoin(self.base_url, url)

    return requests.get(url, headers=headers, timeout=self.timeout)

esiosapy.AsyncESIOSAPYClient

An async client for interacting with the ESIOS API.

This client provides access to various managers that handle specific types of requests to the ESIOS API, such as archives, indicators, and offer indicators. It uses httpx for async HTTP operations.

Requires httpx to be installed. Install with: pip install esiosapy[async]

Source code in esiosapy/async_client.py
class AsyncESIOSAPYClient:
    """
    An async client for interacting with the ESIOS API.

    This client provides access to various managers that handle specific
    types of requests to the ESIOS API, such as archives, indicators, and
    offer indicators. It uses httpx for async HTTP operations.

    Requires httpx to be installed. Install with: pip install esiosapy[async]
    """

    def __init__(
        self,
        token: str,
        base_url: str = ESIOS_API_URL,
        timeout: int = 30,
    ) -> None:
        """
        Initializes the AsyncESIOSAPYClient with an API token and a base URL.

        :param token: The API token used for authentication.
        :type token: str
        :param base_url: The base URL for the ESIOS API. Defaults to ESIOS_API_URL.
        :type base_url: str, optional
        :param timeout: Request timeout in seconds, defaults to 30.
        :type timeout: int
        :raises ImportError: If httpx is not installed.
        """
        if not _HTTPX_AVAILABLE:
            raise ImportError(
                "The `httpx` package is required to use AsyncESIOSAPYClient. "
                "Install it with 'pip install httpx' "
                "or with your preferred package manager."
            )
        self.token = token
        self.base_url = base_url
        self.request_helper = AsyncRequestHelper(base_url, token, timeout)

        self.archives: AsyncArchiveManager = AsyncArchiveManager(self.request_helper)
        self.indicators: AsyncIndicatorManager = AsyncIndicatorManager(
            self.request_helper
        )
        self.offer_indicators: AsyncOfferIndicatorManager = AsyncOfferIndicatorManager(
            self.request_helper
        )

    async def close(self) -> None:
        """Close the async client and release resources."""
        await self.request_helper.close()

    async def __aenter__(self) -> AsyncESIOSAPYClient:
        return self

    async def __aexit__(self, *args: Any) -> None:
        await self.close()

    async def raw_request(
        self,
        url: str,
        headers: dict[str, str] | None = None,
    ) -> httpx.Response:
        """
        Makes a raw async GET request to a specified URL with optional headers.

        This method allows for making a direct GET request to a specified URL.
        It adds default headers to the request and handles URL construction
        if the provided URL is relative.

        :param url: The URL to which the GET request is made. If the URL is
                    relative, it will be joined with the base URL.
        :type url: str
        :param headers: Optional headers to include in the request. Defaults to None.
        :type headers: Optional[Dict[str, str]], optional
        :return: The response object resulting from the GET request.
        :rtype: httpx.Response
        """
        if headers is None:
            headers = {}
        headers = self.request_helper.add_default_headers(headers)

        if urlparse(url).netloc == "":
            url = urljoin(self.base_url, url)

        client = self.request_helper._get_client()
        response = await client.get(
            url, headers=headers, timeout=self.request_helper.timeout
        )
        return response

Functions

__init__(token, base_url=ESIOS_API_URL, timeout=30)

Initializes the AsyncESIOSAPYClient with an API token and a base URL.

Parameters:

Name Type Description Default
token str

The API token used for authentication.

required
base_url str

The base URL for the ESIOS API. Defaults to ESIOS_API_URL.

ESIOS_API_URL
timeout int

Request timeout in seconds, defaults to 30.

30

Raises:

Type Description
ImportError

If httpx is not installed.

Source code in esiosapy/async_client.py
def __init__(
    self,
    token: str,
    base_url: str = ESIOS_API_URL,
    timeout: int = 30,
) -> None:
    """
    Initializes the AsyncESIOSAPYClient with an API token and a base URL.

    :param token: The API token used for authentication.
    :type token: str
    :param base_url: The base URL for the ESIOS API. Defaults to ESIOS_API_URL.
    :type base_url: str, optional
    :param timeout: Request timeout in seconds, defaults to 30.
    :type timeout: int
    :raises ImportError: If httpx is not installed.
    """
    if not _HTTPX_AVAILABLE:
        raise ImportError(
            "The `httpx` package is required to use AsyncESIOSAPYClient. "
            "Install it with 'pip install httpx' "
            "or with your preferred package manager."
        )
    self.token = token
    self.base_url = base_url
    self.request_helper = AsyncRequestHelper(base_url, token, timeout)

    self.archives: AsyncArchiveManager = AsyncArchiveManager(self.request_helper)
    self.indicators: AsyncIndicatorManager = AsyncIndicatorManager(
        self.request_helper
    )
    self.offer_indicators: AsyncOfferIndicatorManager = AsyncOfferIndicatorManager(
        self.request_helper
    )

close() async

Close the async client and release resources.

Source code in esiosapy/async_client.py
async def close(self) -> None:
    """Close the async client and release resources."""
    await self.request_helper.close()

raw_request(url, headers=None) async

Makes a raw async GET request to a specified URL with optional headers.

This method allows for making a direct GET request to a specified URL. It adds default headers to the request and handles URL construction if the provided URL is relative.

Parameters:

Name Type Description Default
url str

The URL to which the GET request is made. If the URL is relative, it will be joined with the base URL.

required
headers dict[str, str] | None

Optional headers to include in the request. Defaults to None.

None

Returns:

Type Description
httpx.Response

The response object resulting from the GET request.

Source code in esiosapy/async_client.py
async def raw_request(
    self,
    url: str,
    headers: dict[str, str] | None = None,
) -> httpx.Response:
    """
    Makes a raw async GET request to a specified URL with optional headers.

    This method allows for making a direct GET request to a specified URL.
    It adds default headers to the request and handles URL construction
    if the provided URL is relative.

    :param url: The URL to which the GET request is made. If the URL is
                relative, it will be joined with the base URL.
    :type url: str
    :param headers: Optional headers to include in the request. Defaults to None.
    :type headers: Optional[Dict[str, str]], optional
    :return: The response object resulting from the GET request.
    :rtype: httpx.Response
    """
    if headers is None:
        headers = {}
    headers = self.request_helper.add_default_headers(headers)

    if urlparse(url).netloc == "":
        url = urljoin(self.base_url, url)

    client = self.request_helper._get_client()
    response = await client.get(
        url, headers=headers, timeout=self.request_helper.timeout
    )
    return response

Managers

esiosapy.IndicatorManager

Bases: BaseIndicatorManager[RequestHelper]

Manages indicator-related operations for the ESIOS API.

This class provides methods to retrieve and search for indicators from the ESIOS API, including listing all available indicators and searching for indicators by name.

Source code in esiosapy/managers/indicator_manager.py
class IndicatorManager(BaseIndicatorManager[RequestHelper]):
    """
    Manages indicator-related operations for the ESIOS API.

    This class provides methods to retrieve and search for indicators from the
    ESIOS API, including listing all available indicators and searching for
    indicators by name.
    """

    def __init__(self, request_helper: RequestHelper) -> None:
        """
        Initializes the IndicatorManager with a RequestHelper.

        :param request_helper: An instance of RequestHelper used to make API requests.
        """
        super().__init__(request_helper)

    def list_all(self, taxonomy_terms: list[str] | None = None) -> list[Indicator]:
        """
        Retrieves a list of all indicators, optionally filtered by taxonomy terms.

        This method sends a GET request to the `/indicators` endpoint and
        returns a list of Indicator objects. If taxonomy terms are provided,
        they are used to filter the indicators.

        :param taxonomy_terms: A list of taxonomy terms to filter the indicators,
                               defaults to None.
        :return: A list of Indicator objects representing all (or filtered) indicators.
        """
        params: dict[str, str | int | list[str]] = {}
        if taxonomy_terms:
            params["taxonomy_terms[]"] = taxonomy_terms

        response = self.request_helper.get_request(self._endpoint, params=params)
        return [
            self._init_indicator(indicator)
            for indicator in response.json()["indicators"]
        ]

    def search(self, name: str) -> list[Indicator]:
        """
        Searches for indicators by name.

        This method sends a GET request to the `/indicators` endpoint with a
        search query, returning a list of Indicator objects that match the
        specified name.

        :param name: The name or part of the name to search for in indicators.
        :return: A list of Indicator objects that match the search query.
        """
        response = self.request_helper.get_request(
            self._endpoint, params={"text": name}
        )
        return [
            self._init_indicator(indicator)
            for indicator in response.json()["indicators"]
        ]

Functions

__init__(request_helper)

Initializes the IndicatorManager with a RequestHelper.

Parameters:

Name Type Description Default
request_helper RequestHelper

An instance of RequestHelper used to make API requests.

required
Source code in esiosapy/managers/indicator_manager.py
def __init__(self, request_helper: RequestHelper) -> None:
    """
    Initializes the IndicatorManager with a RequestHelper.

    :param request_helper: An instance of RequestHelper used to make API requests.
    """
    super().__init__(request_helper)

list_all(taxonomy_terms=None)

Retrieves a list of all indicators, optionally filtered by taxonomy terms.

This method sends a GET request to the /indicators endpoint and returns a list of Indicator objects. If taxonomy terms are provided, they are used to filter the indicators.

Parameters:

Name Type Description Default
taxonomy_terms list[str] | None

A list of taxonomy terms to filter the indicators, defaults to None.

None

Returns:

Type Description
list[Indicator]

A list of Indicator objects representing all (or filtered) indicators.

Source code in esiosapy/managers/indicator_manager.py
def list_all(self, taxonomy_terms: list[str] | None = None) -> list[Indicator]:
    """
    Retrieves a list of all indicators, optionally filtered by taxonomy terms.

    This method sends a GET request to the `/indicators` endpoint and
    returns a list of Indicator objects. If taxonomy terms are provided,
    they are used to filter the indicators.

    :param taxonomy_terms: A list of taxonomy terms to filter the indicators,
                           defaults to None.
    :return: A list of Indicator objects representing all (or filtered) indicators.
    """
    params: dict[str, str | int | list[str]] = {}
    if taxonomy_terms:
        params["taxonomy_terms[]"] = taxonomy_terms

    response = self.request_helper.get_request(self._endpoint, params=params)
    return [
        self._init_indicator(indicator)
        for indicator in response.json()["indicators"]
    ]

search(name)

Searches for indicators by name.

This method sends a GET request to the /indicators endpoint with a search query, returning a list of Indicator objects that match the specified name.

Parameters:

Name Type Description Default
name str

The name or part of the name to search for in indicators.

required

Returns:

Type Description
list[Indicator]

A list of Indicator objects that match the search query.

Source code in esiosapy/managers/indicator_manager.py
def search(self, name: str) -> list[Indicator]:
    """
    Searches for indicators by name.

    This method sends a GET request to the `/indicators` endpoint with a
    search query, returning a list of Indicator objects that match the
    specified name.

    :param name: The name or part of the name to search for in indicators.
    :return: A list of Indicator objects that match the search query.
    """
    response = self.request_helper.get_request(
        self._endpoint, params={"text": name}
    )
    return [
        self._init_indicator(indicator)
        for indicator in response.json()["indicators"]
    ]

esiosapy.AsyncIndicatorManager

Bases: BaseIndicatorManager[AsyncRequestHelper]

Manages indicator-related operations for the ESIOS API (async version).

This class provides methods to retrieve and search for indicators from the ESIOS API, including listing all available indicators and searching for indicators by name.

Source code in esiosapy/managers/async_indicator_manager.py
class AsyncIndicatorManager(BaseIndicatorManager[AsyncRequestHelper]):
    """
    Manages indicator-related operations for the ESIOS API (async version).

    This class provides methods to retrieve and search for indicators from the
    ESIOS API, including listing all available indicators and searching for
    indicators by name.
    """

    def __init__(self, request_helper: AsyncRequestHelper) -> None:
        """
        Initializes the AsyncIndicatorManager with an AsyncRequestHelper.

        :param request_helper: An instance of AsyncRequestHelper used to make API requests.
        """
        super().__init__(request_helper)

    async def list_all(
        self, taxonomy_terms: list[str] | None = None
    ) -> list[Indicator]:
        """
        Retrieves a list of all indicators, optionally filtered by taxonomy terms.

        This method sends a GET request to the `/indicators` endpoint and
        returns a list of Indicator objects.

        :param taxonomy_terms: A list of taxonomy terms to filter the indicators,
                               defaults to None.
        :return: A list of Indicator objects representing all (or filtered) indicators.
        """
        params: dict[str, str | int | list[str]] = {}
        if taxonomy_terms:
            params["taxonomy_terms[]"] = taxonomy_terms

        response = await self.request_helper.get_request(self._endpoint, params=params)
        return [
            self._init_indicator(indicator)
            for indicator in response.json()["indicators"]
        ]

    async def search(self, name: str) -> list[Indicator]:
        """
        Searches for indicators by name.

        This method sends a GET request to the `/indicators` endpoint with a
        search query, returning a list of Indicator objects.

        :param name: The name or part of the name to search for in indicators.
        :return: A list of Indicator objects that match the search query.
        """
        response = await self.request_helper.get_request(
            self._endpoint, params={"text": name}
        )
        return [
            self._init_indicator(indicator)
            for indicator in response.json()["indicators"]
        ]

Functions

__init__(request_helper)

Initializes the AsyncIndicatorManager with an AsyncRequestHelper.

Parameters:

Name Type Description Default
request_helper AsyncRequestHelper

An instance of AsyncRequestHelper used to make API requests.

required
Source code in esiosapy/managers/async_indicator_manager.py
def __init__(self, request_helper: AsyncRequestHelper) -> None:
    """
    Initializes the AsyncIndicatorManager with an AsyncRequestHelper.

    :param request_helper: An instance of AsyncRequestHelper used to make API requests.
    """
    super().__init__(request_helper)

list_all(taxonomy_terms=None) async

Retrieves a list of all indicators, optionally filtered by taxonomy terms.

This method sends a GET request to the /indicators endpoint and returns a list of Indicator objects.

Parameters:

Name Type Description Default
taxonomy_terms list[str] | None

A list of taxonomy terms to filter the indicators, defaults to None.

None

Returns:

Type Description
list[Indicator]

A list of Indicator objects representing all (or filtered) indicators.

Source code in esiosapy/managers/async_indicator_manager.py
async def list_all(
    self, taxonomy_terms: list[str] | None = None
) -> list[Indicator]:
    """
    Retrieves a list of all indicators, optionally filtered by taxonomy terms.

    This method sends a GET request to the `/indicators` endpoint and
    returns a list of Indicator objects.

    :param taxonomy_terms: A list of taxonomy terms to filter the indicators,
                           defaults to None.
    :return: A list of Indicator objects representing all (or filtered) indicators.
    """
    params: dict[str, str | int | list[str]] = {}
    if taxonomy_terms:
        params["taxonomy_terms[]"] = taxonomy_terms

    response = await self.request_helper.get_request(self._endpoint, params=params)
    return [
        self._init_indicator(indicator)
        for indicator in response.json()["indicators"]
    ]

search(name) async

Searches for indicators by name.

This method sends a GET request to the /indicators endpoint with a search query, returning a list of Indicator objects.

Parameters:

Name Type Description Default
name str

The name or part of the name to search for in indicators.

required

Returns:

Type Description
list[Indicator]

A list of Indicator objects that match the search query.

Source code in esiosapy/managers/async_indicator_manager.py
async def search(self, name: str) -> list[Indicator]:
    """
    Searches for indicators by name.

    This method sends a GET request to the `/indicators` endpoint with a
    search query, returning a list of Indicator objects.

    :param name: The name or part of the name to search for in indicators.
    :return: A list of Indicator objects that match the search query.
    """
    response = await self.request_helper.get_request(
        self._endpoint, params={"text": name}
    )
    return [
        self._init_indicator(indicator)
        for indicator in response.json()["indicators"]
    ]

esiosapy.ArchiveManager

Bases: BaseArchiveManager[RequestHelper]

Manages archive-related operations for the ESIOS API.

This class provides methods to retrieve and filter archives from the ESIOS API, such as listing all archives or filtering them by date or date range.

Source code in esiosapy/managers/archive_manager.py
class ArchiveManager(BaseArchiveManager[RequestHelper]):
    """
    Manages archive-related operations for the ESIOS API.

    This class provides methods to retrieve and filter archives from the ESIOS
    API, such as listing all archives or filtering them by date or date range.
    """

    def __init__(self, request_helper: RequestHelper) -> None:
        """
        Initializes the ArchiveManager with a RequestHelper.

        :param request_helper: An instance of RequestHelper used to make API requests.
        """
        super().__init__(request_helper)

    def list_all(self) -> list[Archive]:
        """
        Retrieves a list of all archives.

        This method sends a GET request to the `/archives` endpoint and
        returns a list of Archive objects representing all available archives.

        :return: A list of Archive objects representing all archives.
        """
        response = self.request_helper.get_request(self._endpoint)
        return [self._init_archive(archive) for archive in response.json()["archives"]]

    def list_by_date(
        self,
        target_dt: datetime | str,
        date_type: ArchiveDateType | None = None,
        taxonomy_terms: list[str] | None = None,
    ) -> list[Archive]:
        """
        Retrieves a list of archives filtered by a specific date.

        This method sends a GET request to the `/archives` endpoint with filters
        based on the provided date, date type, and optional taxonomy terms.

        :param target_dt: The target date for filtering archives. Can be a datetime
                          object or an ISO 8601 formatted string.
        :param date_type: The type of date to filter by (e.g., publication date),
                          defaults to None.
        :param taxonomy_terms: A list of taxonomy terms to further filter the archives,
                               defaults to None.
        :return: A list of Archive objects filtered by the specified date.
        """
        if isinstance(target_dt, datetime):
            target_dt = target_dt.strftime("%Y-%m-%dT%H:%M:%S.%f%z")

        params: dict[str, str | int | list[str]] = {"date": target_dt}
        if date_type:
            params["date_type"] = date_type.value
        if taxonomy_terms:
            params["taxonomy_terms[]"] = taxonomy_terms

        response = self.request_helper.get_request(self._endpoint, params=params)
        return [self._init_archive(archive) for archive in response.json()["archives"]]

    def list_by_date_range(
        self,
        target_dt_start: datetime | str,
        target_dt_end: datetime | str,
        date_type: ArchiveDateType | None = None,
        taxonomy_terms: list[str] | None = None,
    ) -> list[Archive]:
        """
        Retrieves a list of archives filtered by a date range.

        This method sends a GET request to the `/archives` endpoint with filters
        based on the provided start and end dates, date type, and optional taxonomy
        terms.

        :param target_dt_start: The start date for filtering archives. Can be a datetime
                                object or an ISO 8601 formatted string.
        :param target_dt_end: The end date for filtering archives. Can be a datetime
                              object or an ISO 8601 formatted string.
        :param date_type: The type of date to filter by (e.g., publication date),
                          defaults to None.
        :param taxonomy_terms: A list of taxonomy terms to further filter the archives,
                               defaults to None.
        :return: A list of Archive objects filtered by the specified date range.
        """
        if isinstance(target_dt_start, datetime):
            target_dt_start = target_dt_start.strftime("%Y-%m-%dT%H:%M:%S.%f%z")
        if isinstance(target_dt_end, datetime):
            target_dt_end = target_dt_end.strftime("%Y-%m-%dT%H:%M:%S.%f%z")

        params: dict[str, str | int | list[str]] = {
            "start_date": target_dt_start,
            "end_date": target_dt_end,
        }
        if date_type:
            params["date_type"] = date_type.value
        if taxonomy_terms:
            params["taxonomy_terms[]"] = taxonomy_terms

        response = self.request_helper.get_request(self._endpoint, params=params)
        return [self._init_archive(archive) for archive in response.json()["archives"]]

Functions

__init__(request_helper)

Initializes the ArchiveManager with a RequestHelper.

Parameters:

Name Type Description Default
request_helper RequestHelper

An instance of RequestHelper used to make API requests.

required
Source code in esiosapy/managers/archive_manager.py
def __init__(self, request_helper: RequestHelper) -> None:
    """
    Initializes the ArchiveManager with a RequestHelper.

    :param request_helper: An instance of RequestHelper used to make API requests.
    """
    super().__init__(request_helper)

list_all()

Retrieves a list of all archives.

This method sends a GET request to the /archives endpoint and returns a list of Archive objects representing all available archives.

Returns:

Type Description
list[Archive]

A list of Archive objects representing all archives.

Source code in esiosapy/managers/archive_manager.py
def list_all(self) -> list[Archive]:
    """
    Retrieves a list of all archives.

    This method sends a GET request to the `/archives` endpoint and
    returns a list of Archive objects representing all available archives.

    :return: A list of Archive objects representing all archives.
    """
    response = self.request_helper.get_request(self._endpoint)
    return [self._init_archive(archive) for archive in response.json()["archives"]]

list_by_date(target_dt, date_type=None, taxonomy_terms=None)

Retrieves a list of archives filtered by a specific date.

This method sends a GET request to the /archives endpoint with filters based on the provided date, date type, and optional taxonomy terms.

Parameters:

Name Type Description Default
target_dt datetime | str

The target date for filtering archives. Can be a datetime object or an ISO 8601 formatted string.

required
date_type ArchiveDateType | None

The type of date to filter by (e.g., publication date), defaults to None.

None
taxonomy_terms list[str] | None

A list of taxonomy terms to further filter the archives, defaults to None.

None

Returns:

Type Description
list[Archive]

A list of Archive objects filtered by the specified date.

Source code in esiosapy/managers/archive_manager.py
def list_by_date(
    self,
    target_dt: datetime | str,
    date_type: ArchiveDateType | None = None,
    taxonomy_terms: list[str] | None = None,
) -> list[Archive]:
    """
    Retrieves a list of archives filtered by a specific date.

    This method sends a GET request to the `/archives` endpoint with filters
    based on the provided date, date type, and optional taxonomy terms.

    :param target_dt: The target date for filtering archives. Can be a datetime
                      object or an ISO 8601 formatted string.
    :param date_type: The type of date to filter by (e.g., publication date),
                      defaults to None.
    :param taxonomy_terms: A list of taxonomy terms to further filter the archives,
                           defaults to None.
    :return: A list of Archive objects filtered by the specified date.
    """
    if isinstance(target_dt, datetime):
        target_dt = target_dt.strftime("%Y-%m-%dT%H:%M:%S.%f%z")

    params: dict[str, str | int | list[str]] = {"date": target_dt}
    if date_type:
        params["date_type"] = date_type.value
    if taxonomy_terms:
        params["taxonomy_terms[]"] = taxonomy_terms

    response = self.request_helper.get_request(self._endpoint, params=params)
    return [self._init_archive(archive) for archive in response.json()["archives"]]

list_by_date_range(target_dt_start, target_dt_end, date_type=None, taxonomy_terms=None)

Retrieves a list of archives filtered by a date range.

This method sends a GET request to the /archives endpoint with filters based on the provided start and end dates, date type, and optional taxonomy terms.

Parameters:

Name Type Description Default
target_dt_start datetime | str

The start date for filtering archives. Can be a datetime object or an ISO 8601 formatted string.

required
target_dt_end datetime | str

The end date for filtering archives. Can be a datetime object or an ISO 8601 formatted string.

required
date_type ArchiveDateType | None

The type of date to filter by (e.g., publication date), defaults to None.

None
taxonomy_terms list[str] | None

A list of taxonomy terms to further filter the archives, defaults to None.

None

Returns:

Type Description
list[Archive]

A list of Archive objects filtered by the specified date range.

Source code in esiosapy/managers/archive_manager.py
def list_by_date_range(
    self,
    target_dt_start: datetime | str,
    target_dt_end: datetime | str,
    date_type: ArchiveDateType | None = None,
    taxonomy_terms: list[str] | None = None,
) -> list[Archive]:
    """
    Retrieves a list of archives filtered by a date range.

    This method sends a GET request to the `/archives` endpoint with filters
    based on the provided start and end dates, date type, and optional taxonomy
    terms.

    :param target_dt_start: The start date for filtering archives. Can be a datetime
                            object or an ISO 8601 formatted string.
    :param target_dt_end: The end date for filtering archives. Can be a datetime
                          object or an ISO 8601 formatted string.
    :param date_type: The type of date to filter by (e.g., publication date),
                      defaults to None.
    :param taxonomy_terms: A list of taxonomy terms to further filter the archives,
                           defaults to None.
    :return: A list of Archive objects filtered by the specified date range.
    """
    if isinstance(target_dt_start, datetime):
        target_dt_start = target_dt_start.strftime("%Y-%m-%dT%H:%M:%S.%f%z")
    if isinstance(target_dt_end, datetime):
        target_dt_end = target_dt_end.strftime("%Y-%m-%dT%H:%M:%S.%f%z")

    params: dict[str, str | int | list[str]] = {
        "start_date": target_dt_start,
        "end_date": target_dt_end,
    }
    if date_type:
        params["date_type"] = date_type.value
    if taxonomy_terms:
        params["taxonomy_terms[]"] = taxonomy_terms

    response = self.request_helper.get_request(self._endpoint, params=params)
    return [self._init_archive(archive) for archive in response.json()["archives"]]

esiosapy.AsyncArchiveManager

Bases: BaseArchiveManager[AsyncRequestHelper]

Manages archive-related operations for the ESIOS API (async version).

This class provides methods to retrieve archives from the ESIOS API, including listing all archives and filtering by date.

Source code in esiosapy/managers/async_archive_manager.py
class AsyncArchiveManager(BaseArchiveManager[AsyncRequestHelper]):
    """
    Manages archive-related operations for the ESIOS API (async version).

    This class provides methods to retrieve archives from the ESIOS API,
    including listing all archives and filtering by date.
    """

    def __init__(self, request_helper: AsyncRequestHelper) -> None:
        """
        Initializes the AsyncArchiveManager with an AsyncRequestHelper.

        :param request_helper: An instance of AsyncRequestHelper used to make API requests.
        """
        super().__init__(request_helper)

    async def list_all(self) -> list[Archive]:
        """
        Retrieves a list of all archives.

        This method sends a GET request to the `/archives` endpoint and
        returns a list of Archive objects representing all available archives.

        :return: A list of Archive objects representing all archives.
        """
        response = await self.request_helper.get_request(self._endpoint)
        return [self._init_archive(archive) for archive in response.json()["archives"]]

    async def list_by_date(
        self,
        target_dt: datetime | str,
        date_type: ArchiveDateType | None = None,
        taxonomy_terms: list[str] | None = None,
    ) -> list[Archive]:
        """
        Retrieves a list of archives filtered by a specific date.

        This method sends a GET request to the `/archives` endpoint with filters
        based on the provided date, date type, and optional taxonomy terms.

        :param target_dt: The target date for filtering archives. Can be a datetime
                          object or an ISO 8601 formatted string.
        :param date_type: The type of date to filter by (e.g., publication date),
                          defaults to None.
        :param taxonomy_terms: A list of taxonomy terms to further filter the archives,
                               defaults to None.
        :return: A list of Archive objects filtered by the specified date.
        """
        if isinstance(target_dt, datetime):
            target_dt = target_dt.strftime("%Y-%m-%dT%H:%M:%S.%f%z")

        params: dict[str, str | int | list[str]] = {"date": target_dt}
        if date_type:
            params["date_type"] = date_type.value
        if taxonomy_terms:
            params["taxonomy_terms[]"] = taxonomy_terms

        response = await self.request_helper.get_request(self._endpoint, params=params)
        return [self._init_archive(archive) for archive in response.json()["archives"]]

    async def list_by_date_range(
        self,
        target_dt_start: datetime | str,
        target_dt_end: datetime | str,
        date_type: ArchiveDateType | None = None,
        taxonomy_terms: list[str] | None = None,
    ) -> list[Archive]:
        """
        Retrieves a list of archives filtered by a date range.

        This method sends a GET request to the `/archives` endpoint with filters
        based on the provided start and end dates, date type, and optional taxonomy
        terms.

        :param target_dt_start: The start date for filtering archives. Can be a datetime
                                object or an ISO 8601 formatted string.
        :param target_dt_end: The end date for filtering archives. Can be a datetime
                              object or an ISO 8601 formatted string.
        :param date_type: The type of date to filter by (e.g., publication date),
                          defaults to None.
        :param taxonomy_terms: A list of taxonomy terms to further filter the archives,
                               defaults to None.
        :return: A list of Archive objects filtered by the specified date range.
        """
        if isinstance(target_dt_start, datetime):
            target_dt_start = target_dt_start.strftime("%Y-%m-%dT%H:%M:%S.%f%z")
        if isinstance(target_dt_end, datetime):
            target_dt_end = target_dt_end.strftime("%Y-%m-%dT%H:%M:%S.%f%z")

        params: dict[str, str | int | list[str]] = {
            "start_date": target_dt_start,
            "end_date": target_dt_end,
        }
        if date_type:
            params["date_type"] = date_type.value
        if taxonomy_terms:
            params["taxonomy_terms[]"] = taxonomy_terms

        response = await self.request_helper.get_request(self._endpoint, params=params)
        return [self._init_archive(archive) for archive in response.json()["archives"]]

Functions

__init__(request_helper)

Initializes the AsyncArchiveManager with an AsyncRequestHelper.

Parameters:

Name Type Description Default
request_helper AsyncRequestHelper

An instance of AsyncRequestHelper used to make API requests.

required
Source code in esiosapy/managers/async_archive_manager.py
def __init__(self, request_helper: AsyncRequestHelper) -> None:
    """
    Initializes the AsyncArchiveManager with an AsyncRequestHelper.

    :param request_helper: An instance of AsyncRequestHelper used to make API requests.
    """
    super().__init__(request_helper)

list_all() async

Retrieves a list of all archives.

This method sends a GET request to the /archives endpoint and returns a list of Archive objects representing all available archives.

Returns:

Type Description
list[Archive]

A list of Archive objects representing all archives.

Source code in esiosapy/managers/async_archive_manager.py
async def list_all(self) -> list[Archive]:
    """
    Retrieves a list of all archives.

    This method sends a GET request to the `/archives` endpoint and
    returns a list of Archive objects representing all available archives.

    :return: A list of Archive objects representing all archives.
    """
    response = await self.request_helper.get_request(self._endpoint)
    return [self._init_archive(archive) for archive in response.json()["archives"]]

list_by_date(target_dt, date_type=None, taxonomy_terms=None) async

Retrieves a list of archives filtered by a specific date.

This method sends a GET request to the /archives endpoint with filters based on the provided date, date type, and optional taxonomy terms.

Parameters:

Name Type Description Default
target_dt datetime | str

The target date for filtering archives. Can be a datetime object or an ISO 8601 formatted string.

required
date_type ArchiveDateType | None

The type of date to filter by (e.g., publication date), defaults to None.

None
taxonomy_terms list[str] | None

A list of taxonomy terms to further filter the archives, defaults to None.

None

Returns:

Type Description
list[Archive]

A list of Archive objects filtered by the specified date.

Source code in esiosapy/managers/async_archive_manager.py
async def list_by_date(
    self,
    target_dt: datetime | str,
    date_type: ArchiveDateType | None = None,
    taxonomy_terms: list[str] | None = None,
) -> list[Archive]:
    """
    Retrieves a list of archives filtered by a specific date.

    This method sends a GET request to the `/archives` endpoint with filters
    based on the provided date, date type, and optional taxonomy terms.

    :param target_dt: The target date for filtering archives. Can be a datetime
                      object or an ISO 8601 formatted string.
    :param date_type: The type of date to filter by (e.g., publication date),
                      defaults to None.
    :param taxonomy_terms: A list of taxonomy terms to further filter the archives,
                           defaults to None.
    :return: A list of Archive objects filtered by the specified date.
    """
    if isinstance(target_dt, datetime):
        target_dt = target_dt.strftime("%Y-%m-%dT%H:%M:%S.%f%z")

    params: dict[str, str | int | list[str]] = {"date": target_dt}
    if date_type:
        params["date_type"] = date_type.value
    if taxonomy_terms:
        params["taxonomy_terms[]"] = taxonomy_terms

    response = await self.request_helper.get_request(self._endpoint, params=params)
    return [self._init_archive(archive) for archive in response.json()["archives"]]

list_by_date_range(target_dt_start, target_dt_end, date_type=None, taxonomy_terms=None) async

Retrieves a list of archives filtered by a date range.

This method sends a GET request to the /archives endpoint with filters based on the provided start and end dates, date type, and optional taxonomy terms.

Parameters:

Name Type Description Default
target_dt_start datetime | str

The start date for filtering archives. Can be a datetime object or an ISO 8601 formatted string.

required
target_dt_end datetime | str

The end date for filtering archives. Can be a datetime object or an ISO 8601 formatted string.

required
date_type ArchiveDateType | None

The type of date to filter by (e.g., publication date), defaults to None.

None
taxonomy_terms list[str] | None

A list of taxonomy terms to further filter the archives, defaults to None.

None

Returns:

Type Description
list[Archive]

A list of Archive objects filtered by the specified date range.

Source code in esiosapy/managers/async_archive_manager.py
async def list_by_date_range(
    self,
    target_dt_start: datetime | str,
    target_dt_end: datetime | str,
    date_type: ArchiveDateType | None = None,
    taxonomy_terms: list[str] | None = None,
) -> list[Archive]:
    """
    Retrieves a list of archives filtered by a date range.

    This method sends a GET request to the `/archives` endpoint with filters
    based on the provided start and end dates, date type, and optional taxonomy
    terms.

    :param target_dt_start: The start date for filtering archives. Can be a datetime
                            object or an ISO 8601 formatted string.
    :param target_dt_end: The end date for filtering archives. Can be a datetime
                          object or an ISO 8601 formatted string.
    :param date_type: The type of date to filter by (e.g., publication date),
                      defaults to None.
    :param taxonomy_terms: A list of taxonomy terms to further filter the archives,
                           defaults to None.
    :return: A list of Archive objects filtered by the specified date range.
    """
    if isinstance(target_dt_start, datetime):
        target_dt_start = target_dt_start.strftime("%Y-%m-%dT%H:%M:%S.%f%z")
    if isinstance(target_dt_end, datetime):
        target_dt_end = target_dt_end.strftime("%Y-%m-%dT%H:%M:%S.%f%z")

    params: dict[str, str | int | list[str]] = {
        "start_date": target_dt_start,
        "end_date": target_dt_end,
    }
    if date_type:
        params["date_type"] = date_type.value
    if taxonomy_terms:
        params["taxonomy_terms[]"] = taxonomy_terms

    response = await self.request_helper.get_request(self._endpoint, params=params)
    return [self._init_archive(archive) for archive in response.json()["archives"]]

esiosapy.OfferIndicatorManager

Bases: BaseOfferIndicatorManager[RequestHelper]

Manages offer indicator-related operations for the ESIOS API.

This class provides methods to retrieve offer indicators from the ESIOS API, including listing all available offer indicators with optional filtering by taxonomy terms.

Source code in esiosapy/managers/offer_indicator_manager.py
class OfferIndicatorManager(BaseOfferIndicatorManager[RequestHelper]):
    """
    Manages offer indicator-related operations for the ESIOS API.

    This class provides methods to retrieve offer indicators from the ESIOS API,
    including listing all available offer indicators with optional filtering by
    taxonomy terms.
    """

    def __init__(self, request_helper: RequestHelper) -> None:
        """
        Initializes the OfferIndicatorManager with a RequestHelper.

        :param request_helper: An instance of RequestHelper used to make API requests.
        """
        super().__init__(request_helper)

    def list_all(self, taxonomy_terms: list[str] | None = None) -> list[OfferIndicator]:
        """
        Retrieves a list of all offer indicators, optionally filtered by taxonomy terms.

        This method sends a GET request to the `/offer_indicators` endpoint and
        returns a list of OfferIndicator objects. If taxonomy terms are provided,
        they are used to filter the offer indicators.

        :param taxonomy_terms: A list of taxonomy terms to filter the offer indicators,
                               defaults to None.
        :return: A list of OfferIndicator objects representing all (or filtered)
                 offer indicators.
        """
        params: dict[str, str | int | list[str]] = {}
        if taxonomy_terms:
            params["taxonomy_terms[]"] = taxonomy_terms

        response = self.request_helper.get_request(self._endpoint, params=params)

        return [
            self._init_indicator(indicator)
            for indicator in response.json()["indicators"]
        ]

Functions

__init__(request_helper)

Initializes the OfferIndicatorManager with a RequestHelper.

Parameters:

Name Type Description Default
request_helper RequestHelper

An instance of RequestHelper used to make API requests.

required
Source code in esiosapy/managers/offer_indicator_manager.py
def __init__(self, request_helper: RequestHelper) -> None:
    """
    Initializes the OfferIndicatorManager with a RequestHelper.

    :param request_helper: An instance of RequestHelper used to make API requests.
    """
    super().__init__(request_helper)

list_all(taxonomy_terms=None)

Retrieves a list of all offer indicators, optionally filtered by taxonomy terms.

This method sends a GET request to the /offer_indicators endpoint and returns a list of OfferIndicator objects. If taxonomy terms are provided, they are used to filter the offer indicators.

Parameters:

Name Type Description Default
taxonomy_terms list[str] | None

A list of taxonomy terms to filter the offer indicators, defaults to None.

None

Returns:

Type Description
list[OfferIndicator]

A list of OfferIndicator objects representing all (or filtered) offer indicators.

Source code in esiosapy/managers/offer_indicator_manager.py
def list_all(self, taxonomy_terms: list[str] | None = None) -> list[OfferIndicator]:
    """
    Retrieves a list of all offer indicators, optionally filtered by taxonomy terms.

    This method sends a GET request to the `/offer_indicators` endpoint and
    returns a list of OfferIndicator objects. If taxonomy terms are provided,
    they are used to filter the offer indicators.

    :param taxonomy_terms: A list of taxonomy terms to filter the offer indicators,
                           defaults to None.
    :return: A list of OfferIndicator objects representing all (or filtered)
             offer indicators.
    """
    params: dict[str, str | int | list[str]] = {}
    if taxonomy_terms:
        params["taxonomy_terms[]"] = taxonomy_terms

    response = self.request_helper.get_request(self._endpoint, params=params)

    return [
        self._init_indicator(indicator)
        for indicator in response.json()["indicators"]
    ]

esiosapy.AsyncOfferIndicatorManager

Bases: BaseOfferIndicatorManager[AsyncRequestHelper]

Manages offer indicator-related operations for the ESIOS API (async version).

This class provides methods to retrieve offer indicators from the ESIOS API, including listing all available offer indicators with optional filtering by taxonomy terms.

Source code in esiosapy/managers/async_offer_indicator_manager.py
class AsyncOfferIndicatorManager(BaseOfferIndicatorManager[AsyncRequestHelper]):
    """
    Manages offer indicator-related operations for the ESIOS API (async version).

    This class provides methods to retrieve offer indicators from the ESIOS API,
    including listing all available offer indicators with optional filtering by
    taxonomy terms.
    """

    def __init__(self, request_helper: AsyncRequestHelper) -> None:
        """
        Initializes the AsyncOfferIndicatorManager with an AsyncRequestHelper.

        :param request_helper: An instance of AsyncRequestHelper used to make API requests.
        """
        super().__init__(request_helper)

    async def list_all(
        self, taxonomy_terms: list[str] | None = None
    ) -> list[OfferIndicator]:
        """
        Retrieves a list of all offer indicators, optionally filtered by taxonomy terms.

        This method sends a GET request to the `/offer_indicators` endpoint and
        returns a list of OfferIndicator objects.

        :param taxonomy_terms: A list of taxonomy terms to filter the offer indicators,
                               defaults to None.
        :return: A list of OfferIndicator objects representing all (or filtered)
                 offer indicators.
        """
        params: dict[str, str | int | list[str]] = {}
        if taxonomy_terms:
            params["taxonomy_terms[]"] = taxonomy_terms

        response = await self.request_helper.get_request(self._endpoint, params=params)

        return [
            self._init_indicator(indicator)
            for indicator in response.json()["indicators"]
        ]

Functions

__init__(request_helper)

Initializes the AsyncOfferIndicatorManager with an AsyncRequestHelper.

Parameters:

Name Type Description Default
request_helper AsyncRequestHelper

An instance of AsyncRequestHelper used to make API requests.

required
Source code in esiosapy/managers/async_offer_indicator_manager.py
def __init__(self, request_helper: AsyncRequestHelper) -> None:
    """
    Initializes the AsyncOfferIndicatorManager with an AsyncRequestHelper.

    :param request_helper: An instance of AsyncRequestHelper used to make API requests.
    """
    super().__init__(request_helper)

list_all(taxonomy_terms=None) async

Retrieves a list of all offer indicators, optionally filtered by taxonomy terms.

This method sends a GET request to the /offer_indicators endpoint and returns a list of OfferIndicator objects.

Parameters:

Name Type Description Default
taxonomy_terms list[str] | None

A list of taxonomy terms to filter the offer indicators, defaults to None.

None

Returns:

Type Description
list[OfferIndicator]

A list of OfferIndicator objects representing all (or filtered) offer indicators.

Source code in esiosapy/managers/async_offer_indicator_manager.py
async def list_all(
    self, taxonomy_terms: list[str] | None = None
) -> list[OfferIndicator]:
    """
    Retrieves a list of all offer indicators, optionally filtered by taxonomy terms.

    This method sends a GET request to the `/offer_indicators` endpoint and
    returns a list of OfferIndicator objects.

    :param taxonomy_terms: A list of taxonomy terms to filter the offer indicators,
                           defaults to None.
    :return: A list of OfferIndicator objects representing all (or filtered)
             offer indicators.
    """
    params: dict[str, str | int | list[str]] = {}
    if taxonomy_terms:
        params["taxonomy_terms[]"] = taxonomy_terms

    response = await self.request_helper.get_request(self._endpoint, params=params)

    return [
        self._init_indicator(indicator)
        for indicator in response.json()["indicators"]
    ]

Exceptions

esiosapy.exceptions.ESIOSAPIError

Bases: Exception

Base exception for all ESIOS API errors.

Source code in esiosapy/exceptions.py
class ESIOSAPIError(Exception):
    """Base exception for all ESIOS API errors."""

    def __init__(self, message: str, details: dict[str, Any] | None = None):
        super().__init__(message)
        self.message = message
        self.details = details or {}

esiosapy.exceptions.APIResponseError

Bases: ESIOSAPIError

Raised when the API returns an error response.

Source code in esiosapy/exceptions.py
class APIResponseError(ESIOSAPIError):
    """Raised when the API returns an error response."""

    def __init__(
        self,
        message: str,
        status_code: int | None = None,
        response_body: str | None = None,
    ):
        super().__init__(
            message, {"status_code": status_code, "response_body": response_body}
        )
        self.status_code = status_code
        self.response_body = response_body

esiosapy.exceptions.AuthenticationError

Bases: ESIOSAPIError

Raised when authentication fails (invalid or missing API token).

Source code in esiosapy/exceptions.py
class AuthenticationError(ESIOSAPIError):
    """Raised when authentication fails (invalid or missing API token)."""

Constants

esiosapy.constants.ESIOS_API_URL = 'https://api.esios.ree.es/' module-attribute

Models

esiosapy.models.Indicator

Bases: BaseModel

Represents an indicator with associated metadata and methods to retrieve and process its data.

This class models an indicator object, which includes various attributes such as its ID, name, description, and raw data. It also provides methods to prettify the description and to fetch the indicator's data over specified date ranges with optional geographical and time-based aggregations and truncations.

Source code in esiosapy/models/indicator/indicator.py
class Indicator(BaseModel):
    """
    Represents an indicator with associated metadata and methods to retrieve
    and process its data.

    This class models an indicator object, which includes various attributes
    such as its ID, name,
    description, and raw data. It also provides methods to prettify
    the description and to fetch
    the indicator's data over specified date ranges with optional
    geographical and time-based
    aggregations and truncations.
    """

    id: int
    """The unique identifier for the indicator.

    :type: int
    """

    name: str
    """The name of the indicator.

    :type: str
    """

    short_name: str
    """The short name of the indicator.

    :type: str
    """

    description: str = "No description available."
    """A detailed description of the indicator.

    :type: str
    """

    raw: dict[str, Any]
    """The raw dictionary containing the original data of the indicator.

    :type: Dict[str, Any]
    """

    _request_helper: RequestHelper
    """An instance of RequestHelper used to make API requests.

    :type: RequestHelper
    """

    def __init__(self, **data: Any):
        """
        Initializes the Indicator class with the provided data.

        :param data: The data used to initialize the indicator, including the
                     request helper and other attributes.
        :type data: Any
        """
        super().__init__(**data)
        self._request_helper = data["_request_helper"]

    def prettify_description(self) -> str:
        """
        Converts the HTML description of the indicator into a plain text format
        with better readability.

        :raises ImportError: If `beautifulsoup4` is not installed.
        :return: The prettified description as a plain text string.
        :rtype: str
        """
        try:
            from bs4 import BeautifulSoup
        except ImportError:
            raise ImportError(
                "The `beautifulsoup4` package is required to prettify the description. "
                "Install it with 'pip install esiosapy[beautifulsoup]' "
                "or 'pip install esiosapy[all]'."
            ) from None

        soup = BeautifulSoup(self.description, "html.parser")
        text = soup.get_text(separator="\n").strip()

        return str(text)

    def get_data(
        self,
        target_dt_start: datetime | str,
        target_dt_end: datetime | str,
        geo_ids: list[str] | None = None,
        geo_agg: GeoAgg | None = None,
        geo_trunc: GeoTrunc | None = None,
        time_agg: TimeAgg | None = None,
        time_trunc: TimeTrunc | None = None,
        all_raw_data: bool = False,
    ) -> Any:
        """
        Retrieves the data for the indicator based on the specified parameters.

        :param target_dt_start: The start date and time for data retrieval.
        :type target_dt_start: Union[datetime, str]
        :param target_dt_end: The end date and time for data retrieval.
        :type target_dt_end: Union[datetime, str]
        :param geo_ids: A list of geographical identifiers
                        to filter data, defaults to None.
        :type geo_ids: Optional[List[str]], optional
        :param geo_agg: The geographical aggregation method, defaults to None.
        :type geo_agg: Optional[GeoAgg], optional
        :param geo_trunc: The geographical truncation level, defaults to None.
        :type geo_trunc: Optional[GeoTrunc], optional
        :param time_agg: The time aggregation method, defaults to None.
        :type time_agg: Optional[TimeAgg], optional
        :param time_trunc: The time truncation level, defaults to None.
        :type time_trunc: Optional[TimeTrunc], optional
        :param all_raw_data: Whether to return all raw data or just
                             the indicator values, defaults to False.
        :type all_raw_data: bool, optional
        :return: The retrieved indicator data, either as raw JSON or processed values.
        :rtype: Any
        """
        if isinstance(target_dt_start, datetime):
            target_dt_start = target_dt_start.strftime("%Y-%m-%dT%H:%M:%S.%f%z")
        if isinstance(target_dt_end, datetime):
            target_dt_end = target_dt_end.strftime("%Y-%m-%dT%H:%M:%S.%f%z")

        params: dict[str, str | int | list[str] | None] = {
            "start_date": target_dt_start,
            "end_date": target_dt_end,
            "geo_ids": ",".join(geo_ids) if geo_ids else None,
            "geo_agg": geo_agg.value if geo_agg else None,
            "geo_trunc": geo_trunc.value if geo_trunc else None,
            "time_agg": time_agg.value if time_agg else None,
            "time_trunc": time_trunc.value if time_trunc else None,
        }
        clean_params = {k: v for k, v in params.items() if v is not None}

        response = self._request_helper.get_request(
            f"/indicators/{self.id}", params=clean_params
        )

        return (
            response.json() if all_raw_data else response.json()["indicator"]["values"]
        )

Attributes

description = 'No description available.' class-attribute instance-attribute

A detailed description of the indicator.

id instance-attribute

The unique identifier for the indicator.

name instance-attribute

The name of the indicator.

raw instance-attribute

The raw dictionary containing the original data of the indicator.

short_name instance-attribute

The short name of the indicator.

Functions

__init__(**data)

Initializes the Indicator class with the provided data.

Parameters:

Name Type Description Default
data Any

The data used to initialize the indicator, including the request helper and other attributes.

{}
Source code in esiosapy/models/indicator/indicator.py
def __init__(self, **data: Any):
    """
    Initializes the Indicator class with the provided data.

    :param data: The data used to initialize the indicator, including the
                 request helper and other attributes.
    :type data: Any
    """
    super().__init__(**data)
    self._request_helper = data["_request_helper"]

get_data(target_dt_start, target_dt_end, geo_ids=None, geo_agg=None, geo_trunc=None, time_agg=None, time_trunc=None, all_raw_data=False)

Retrieves the data for the indicator based on the specified parameters.

Parameters:

Name Type Description Default
target_dt_start datetime | str

The start date and time for data retrieval.

required
target_dt_end datetime | str

The end date and time for data retrieval.

required
geo_ids list[str] | None

A list of geographical identifiers to filter data, defaults to None.

None
geo_agg GeoAgg | None

The geographical aggregation method, defaults to None.

None
geo_trunc GeoTrunc | None

The geographical truncation level, defaults to None.

None
time_agg TimeAgg | None

The time aggregation method, defaults to None.

None
time_trunc TimeTrunc | None

The time truncation level, defaults to None.

None
all_raw_data bool

Whether to return all raw data or just the indicator values, defaults to False.

False

Returns:

Type Description
Any

The retrieved indicator data, either as raw JSON or processed values.

Source code in esiosapy/models/indicator/indicator.py
def get_data(
    self,
    target_dt_start: datetime | str,
    target_dt_end: datetime | str,
    geo_ids: list[str] | None = None,
    geo_agg: GeoAgg | None = None,
    geo_trunc: GeoTrunc | None = None,
    time_agg: TimeAgg | None = None,
    time_trunc: TimeTrunc | None = None,
    all_raw_data: bool = False,
) -> Any:
    """
    Retrieves the data for the indicator based on the specified parameters.

    :param target_dt_start: The start date and time for data retrieval.
    :type target_dt_start: Union[datetime, str]
    :param target_dt_end: The end date and time for data retrieval.
    :type target_dt_end: Union[datetime, str]
    :param geo_ids: A list of geographical identifiers
                    to filter data, defaults to None.
    :type geo_ids: Optional[List[str]], optional
    :param geo_agg: The geographical aggregation method, defaults to None.
    :type geo_agg: Optional[GeoAgg], optional
    :param geo_trunc: The geographical truncation level, defaults to None.
    :type geo_trunc: Optional[GeoTrunc], optional
    :param time_agg: The time aggregation method, defaults to None.
    :type time_agg: Optional[TimeAgg], optional
    :param time_trunc: The time truncation level, defaults to None.
    :type time_trunc: Optional[TimeTrunc], optional
    :param all_raw_data: Whether to return all raw data or just
                         the indicator values, defaults to False.
    :type all_raw_data: bool, optional
    :return: The retrieved indicator data, either as raw JSON or processed values.
    :rtype: Any
    """
    if isinstance(target_dt_start, datetime):
        target_dt_start = target_dt_start.strftime("%Y-%m-%dT%H:%M:%S.%f%z")
    if isinstance(target_dt_end, datetime):
        target_dt_end = target_dt_end.strftime("%Y-%m-%dT%H:%M:%S.%f%z")

    params: dict[str, str | int | list[str] | None] = {
        "start_date": target_dt_start,
        "end_date": target_dt_end,
        "geo_ids": ",".join(geo_ids) if geo_ids else None,
        "geo_agg": geo_agg.value if geo_agg else None,
        "geo_trunc": geo_trunc.value if geo_trunc else None,
        "time_agg": time_agg.value if time_agg else None,
        "time_trunc": time_trunc.value if time_trunc else None,
    }
    clean_params = {k: v for k, v in params.items() if v is not None}

    response = self._request_helper.get_request(
        f"/indicators/{self.id}", params=clean_params
    )

    return (
        response.json() if all_raw_data else response.json()["indicator"]["values"]
    )

prettify_description()

Converts the HTML description of the indicator into a plain text format with better readability.

Returns:

Type Description
str

The prettified description as a plain text string.

Raises:

Type Description
ImportError

If beautifulsoup4 is not installed.

Source code in esiosapy/models/indicator/indicator.py
def prettify_description(self) -> str:
    """
    Converts the HTML description of the indicator into a plain text format
    with better readability.

    :raises ImportError: If `beautifulsoup4` is not installed.
    :return: The prettified description as a plain text string.
    :rtype: str
    """
    try:
        from bs4 import BeautifulSoup
    except ImportError:
        raise ImportError(
            "The `beautifulsoup4` package is required to prettify the description. "
            "Install it with 'pip install esiosapy[beautifulsoup]' "
            "or 'pip install esiosapy[all]'."
        ) from None

    soup = BeautifulSoup(self.description, "html.parser")
    text = soup.get_text(separator="\n").strip()

    return str(text)

esiosapy.models.Archive

Bases: BaseModel

Represents an archive with metadata and methods to download associated files.

This class models an archive object with various attributes such as its ID, name, type, and associated download information. It also provides a method to download the archive file, with options to unzip the file and remove the original zip file.

Source code in esiosapy/models/archive/archive.py
class Archive(BaseModel):
    """
    Represents an archive with metadata and methods to download associated files.

    This class models an archive object with various attributes such as its ID, name,
    type, and associated download information. It also provides a method to download
    the archive file, with options to unzip the file and remove the original zip file.
    """

    id: int
    """The unique identifier for the archive.

    :type: int
    """

    name: str
    """The name of the archive.

    :type: str
    """

    horizon: str
    """The horizon associated with the archive.

    :type: str
    """

    archive_type: str
    """The type/category of the archive.

    :type: str
    """

    download: ArchiveDownload
    """The download information associated with the archive.

    :type: ArchiveDownload
    """

    _date: datetime
    """The date associated with the archive.

    :type: datetime
    """

    date_times: list[date] = Field(default_factory=list)
    """A list of dates associated with the archive.

    :type: List[date]
    """

    publication_date: list[date] = Field(default_factory=list)
    """A list of publication dates for the archive.

    :type: List[date]
    """

    taxonomy_terms: list[TaxonomyTerm] = Field(default_factory=list)
    """A list of taxonomy terms associated with the archive.

    :type: List[TaxonomyTerm]
    """

    vocabularies: list[Vocabulary] = Field(default_factory=list)
    """A list of vocabularies associated with the archive.

    :type: List[Vocabulary]
    """

    raw: dict[str, Any]
    """The raw data from which the archive object was created.

    :type: Dict[str, Any]
    """

    _request_helper: RequestHelper
    """An instance of RequestHelper used to make API requests.

    :type: RequestHelper
    """

    def __init__(self, **data: Any):
        """
        Initializes the Archive object with the provided data.

        :param data: The data used to initialize the Archive object. This includes
                     all the attributes as well as a RequestHelper instance
                     for making API requests.
        :type data: Any
        """
        super().__init__(**data)
        self._request_helper = data["_request_helper"]

    def download_file(
        self,
        path: str | Path | None = None,
        unzip: bool = True,
        remove_zip: bool = True,
    ) -> None:
        """
        Downloads the archive file and optionally unzips it.

        This method downloads the file associated with the archive to the specified
        path. The file can be automatically unzipped and the original zip file can
        be removed based on the provided options.

        :param path: The directory where the file should be downloaded. If not provided,
                     the current working directory is used.
        :type path: Optional[Union[str, Path]], optional
        :param unzip: Whether to unzip the downloaded file, defaults to True.
        :type unzip: bool, optional
        :param remove_zip: Whether to remove the original zip file after unzipping,
                           defaults to True.
        :type remove_zip: bool, optional
        :return: None
        """
        if path is None:
            path = Path.cwd()

        response = self._request_helper.get_request(self.download.url)

        zip_path = Path(os.path.join(path, f"{self.name}.zip"))

        with open(zip_path, "wb") as f:
            f.write(response.content)

        if unzip:
            recursive_unzip(zip_path, zip_path.stem, remove=remove_zip)

Attributes

archive_type instance-attribute

The type/category of the archive.

date_times = Field(default_factory=list) class-attribute instance-attribute

A list of dates associated with the archive.

download instance-attribute

The download information associated with the archive.

horizon instance-attribute

The horizon associated with the archive.

id instance-attribute

The unique identifier for the archive.

name instance-attribute

The name of the archive.

publication_date = Field(default_factory=list) class-attribute instance-attribute

A list of publication dates for the archive.

raw instance-attribute

The raw data from which the archive object was created.

taxonomy_terms = Field(default_factory=list) class-attribute instance-attribute

A list of taxonomy terms associated with the archive.

vocabularies = Field(default_factory=list) class-attribute instance-attribute

A list of vocabularies associated with the archive.

Functions

__init__(**data)

Initializes the Archive object with the provided data.

Parameters:

Name Type Description Default
data Any

The data used to initialize the Archive object. This includes all the attributes as well as a RequestHelper instance for making API requests.

{}
Source code in esiosapy/models/archive/archive.py
def __init__(self, **data: Any):
    """
    Initializes the Archive object with the provided data.

    :param data: The data used to initialize the Archive object. This includes
                 all the attributes as well as a RequestHelper instance
                 for making API requests.
    :type data: Any
    """
    super().__init__(**data)
    self._request_helper = data["_request_helper"]

download_file(path=None, unzip=True, remove_zip=True)

Downloads the archive file and optionally unzips it.

This method downloads the file associated with the archive to the specified path. The file can be automatically unzipped and the original zip file can be removed based on the provided options.

Parameters:

Name Type Description Default
path str | Path | None

The directory where the file should be downloaded. If not provided, the current working directory is used.

None
unzip bool

Whether to unzip the downloaded file, defaults to True.

True
remove_zip bool

Whether to remove the original zip file after unzipping, defaults to True.

True

Returns:

Type Description
None

None

Source code in esiosapy/models/archive/archive.py
def download_file(
    self,
    path: str | Path | None = None,
    unzip: bool = True,
    remove_zip: bool = True,
) -> None:
    """
    Downloads the archive file and optionally unzips it.

    This method downloads the file associated with the archive to the specified
    path. The file can be automatically unzipped and the original zip file can
    be removed based on the provided options.

    :param path: The directory where the file should be downloaded. If not provided,
                 the current working directory is used.
    :type path: Optional[Union[str, Path]], optional
    :param unzip: Whether to unzip the downloaded file, defaults to True.
    :type unzip: bool, optional
    :param remove_zip: Whether to remove the original zip file after unzipping,
                       defaults to True.
    :type remove_zip: bool, optional
    :return: None
    """
    if path is None:
        path = Path.cwd()

    response = self._request_helper.get_request(self.download.url)

    zip_path = Path(os.path.join(path, f"{self.name}.zip"))

    with open(zip_path, "wb") as f:
        f.write(response.content)

    if unzip:
        recursive_unzip(zip_path, zip_path.stem, remove=remove_zip)

esiosapy.models.OfferIndicator

Bases: BaseModel

Source code in esiosapy/models/offer_indicator/offer_indicator.py
class OfferIndicator(BaseModel):
    id: int
    """The unique identifier of the offer indicator.

    :type: int
    """

    name: str
    """The name of the offer indicator.

    :type: str
    """

    description: str
    """A detailed description of the offer indicator, often in HTML format.

    :type: str
    """

    raw: dict[str, Any]
    """Raw data associated with the offer indicator.

    :type: Dict[str, Any]
    """

    _request_helper: RequestHelper
    """A helper object for making HTTP requests.

    :type: RequestHelper
    """

    def __init__(self, **data: Any):
        """
        Initialize the OfferIndicator instance.

        :param data: Arbitrary keyword arguments that initialize the object.
        :type data: Any
        """
        super().__init__(**data)
        self._request_helper = data["_request_helper"]

    def prettify_description(self) -> str:
        """
        Convert the HTML description into a prettified plain-text format.

        This method uses BeautifulSoup to parse and clean the HTML content
        found in the description, returning it as a plain-text string.

        :return: A prettified plain-text version of the description.
        :rtype: str

        :raises ImportError: If the BeautifulSoup package is not installed.
        """
        try:
            from bs4 import BeautifulSoup
        except ImportError:
            raise ImportError(
                "The `beautifulsoup4` package is required to prettify the description. "
                "Install it with 'pip install esiosapy[beautifulsoup]' "
                "or 'pip install esiosapy[all]'."
            ) from None

        soup = BeautifulSoup(self.description, "html.parser")
        text = soup.get_text(separator="\n").strip()

        return str(text)

    def get_data_by_date(
        self,
        target_dt: datetime | str,
        all_raw_data: bool = False,
    ) -> Any:
        """
        Retrieve the indicator data for a specific date.

        This method fetches the indicator data for a given date, either returning
        the raw JSON response or the specific indicator values.

        :param target_dt: The target date for which to retrieve data,
                          either as a datetime object or a string.
        :type target_dt: Union[datetime, str]
        :param all_raw_data: If True, returns the entire raw JSON response; otherwise,
                             only returns the indicator values.
        :type all_raw_data: bool, optional
        :return: The requested data, either as a raw JSON or
                 as specific indicator values.
        :rtype: Any
        """
        if isinstance(target_dt, datetime):
            target_dt = target_dt.strftime("%Y-%m-%dT%H:%M:%S.%f%z")

        params: dict[str, str | int | list[str]] = {
            "datetime": target_dt,
        }

        response = self._request_helper.get_request(
            f"/offer_indicators/{self.id}", params=params
        )

        return (
            response.json() if all_raw_data else response.json()["indicator"]["values"]
        )

    def get_data_by_date_range(
        self,
        target_dt_start: datetime | str,
        target_dt_end: datetime | str,
        all_raw_data: bool = False,
    ) -> Any:
        """
        Retrieve the indicator data for a specific date range.

        This method fetches the indicator data for a given date range, either returning
        the raw JSON response or the specific indicator values.

        :param target_dt_start: The start date for the range,
                                either as a datetime object or a string.
        :type target_dt_start: Union[datetime, str]
        :param target_dt_end: The end date for the range,
                              either as a datetime object or a string.
        :type target_dt_end: Union[datetime, str]
        :param all_raw_data: If True, returns the entire raw JSON response; otherwise,
                             only returns the indicator values.
        :type all_raw_data: bool, optional
        :return: The requested data, either as a raw JSON or
                 as specific indicator values.
        :rtype: Any
        """
        if isinstance(target_dt_start, datetime):
            target_dt_start = target_dt_start.strftime("%Y-%m-%dT%H:%M:%S.%f%z")
        if isinstance(target_dt_end, datetime):
            target_dt_end = target_dt_end.strftime("%Y-%m-%dT%H:%M:%S.%f%z")

        params: dict[str, str | int | list[str]] = {
            "start_date": target_dt_start,
            "end_date": target_dt_end,
        }

        response = self._request_helper.get_request(
            f"/offer_indicators/{self.id}", params=params
        )

        return (
            response.json() if all_raw_data else response.json()["indicator"]["values"]
        )

Attributes

description instance-attribute

A detailed description of the offer indicator, often in HTML format.

id instance-attribute

The unique identifier of the offer indicator.

name instance-attribute

The name of the offer indicator.

raw instance-attribute

Raw data associated with the offer indicator.

Functions

__init__(**data)

Initialize the OfferIndicator instance.

Parameters:

Name Type Description Default
data Any

Arbitrary keyword arguments that initialize the object.

{}
Source code in esiosapy/models/offer_indicator/offer_indicator.py
def __init__(self, **data: Any):
    """
    Initialize the OfferIndicator instance.

    :param data: Arbitrary keyword arguments that initialize the object.
    :type data: Any
    """
    super().__init__(**data)
    self._request_helper = data["_request_helper"]

get_data_by_date(target_dt, all_raw_data=False)

Retrieve the indicator data for a specific date.

This method fetches the indicator data for a given date, either returning the raw JSON response or the specific indicator values.

Parameters:

Name Type Description Default
target_dt datetime | str

The target date for which to retrieve data, either as a datetime object or a string.

required
all_raw_data bool

If True, returns the entire raw JSON response; otherwise, only returns the indicator values.

False

Returns:

Type Description
Any

The requested data, either as a raw JSON or as specific indicator values.

Source code in esiosapy/models/offer_indicator/offer_indicator.py
def get_data_by_date(
    self,
    target_dt: datetime | str,
    all_raw_data: bool = False,
) -> Any:
    """
    Retrieve the indicator data for a specific date.

    This method fetches the indicator data for a given date, either returning
    the raw JSON response or the specific indicator values.

    :param target_dt: The target date for which to retrieve data,
                      either as a datetime object or a string.
    :type target_dt: Union[datetime, str]
    :param all_raw_data: If True, returns the entire raw JSON response; otherwise,
                         only returns the indicator values.
    :type all_raw_data: bool, optional
    :return: The requested data, either as a raw JSON or
             as specific indicator values.
    :rtype: Any
    """
    if isinstance(target_dt, datetime):
        target_dt = target_dt.strftime("%Y-%m-%dT%H:%M:%S.%f%z")

    params: dict[str, str | int | list[str]] = {
        "datetime": target_dt,
    }

    response = self._request_helper.get_request(
        f"/offer_indicators/{self.id}", params=params
    )

    return (
        response.json() if all_raw_data else response.json()["indicator"]["values"]
    )

get_data_by_date_range(target_dt_start, target_dt_end, all_raw_data=False)

Retrieve the indicator data for a specific date range.

This method fetches the indicator data for a given date range, either returning the raw JSON response or the specific indicator values.

Parameters:

Name Type Description Default
target_dt_start datetime | str

The start date for the range, either as a datetime object or a string.

required
target_dt_end datetime | str

The end date for the range, either as a datetime object or a string.

required
all_raw_data bool

If True, returns the entire raw JSON response; otherwise, only returns the indicator values.

False

Returns:

Type Description
Any

The requested data, either as a raw JSON or as specific indicator values.

Source code in esiosapy/models/offer_indicator/offer_indicator.py
def get_data_by_date_range(
    self,
    target_dt_start: datetime | str,
    target_dt_end: datetime | str,
    all_raw_data: bool = False,
) -> Any:
    """
    Retrieve the indicator data for a specific date range.

    This method fetches the indicator data for a given date range, either returning
    the raw JSON response or the specific indicator values.

    :param target_dt_start: The start date for the range,
                            either as a datetime object or a string.
    :type target_dt_start: Union[datetime, str]
    :param target_dt_end: The end date for the range,
                          either as a datetime object or a string.
    :type target_dt_end: Union[datetime, str]
    :param all_raw_data: If True, returns the entire raw JSON response; otherwise,
                         only returns the indicator values.
    :type all_raw_data: bool, optional
    :return: The requested data, either as a raw JSON or
             as specific indicator values.
    :rtype: Any
    """
    if isinstance(target_dt_start, datetime):
        target_dt_start = target_dt_start.strftime("%Y-%m-%dT%H:%M:%S.%f%z")
    if isinstance(target_dt_end, datetime):
        target_dt_end = target_dt_end.strftime("%Y-%m-%dT%H:%M:%S.%f%z")

    params: dict[str, str | int | list[str]] = {
        "start_date": target_dt_start,
        "end_date": target_dt_end,
    }

    response = self._request_helper.get_request(
        f"/offer_indicators/{self.id}", params=params
    )

    return (
        response.json() if all_raw_data else response.json()["indicator"]["values"]
    )

prettify_description()

Convert the HTML description into a prettified plain-text format.

This method uses BeautifulSoup to parse and clean the HTML content found in the description, returning it as a plain-text string.

Returns:

Type Description
str

A prettified plain-text version of the description.

Raises:

Type Description
ImportError

If the BeautifulSoup package is not installed.

Source code in esiosapy/models/offer_indicator/offer_indicator.py
def prettify_description(self) -> str:
    """
    Convert the HTML description into a prettified plain-text format.

    This method uses BeautifulSoup to parse and clean the HTML content
    found in the description, returning it as a plain-text string.

    :return: A prettified plain-text version of the description.
    :rtype: str

    :raises ImportError: If the BeautifulSoup package is not installed.
    """
    try:
        from bs4 import BeautifulSoup
    except ImportError:
        raise ImportError(
            "The `beautifulsoup4` package is required to prettify the description. "
            "Install it with 'pip install esiosapy[beautifulsoup]' "
            "or 'pip install esiosapy[all]'."
        ) from None

    soup = BeautifulSoup(self.description, "html.parser")
    text = soup.get_text(separator="\n").strip()

    return str(text)