Skip to content

Client

client

HTTP client for the OpenFIGI REST API.

OpenFIGIClient(api_key: str | None = None, timeout: float = 30.0)

Client for the OpenFIGI mapping REST API.

Source code in src/equities_classifier/clients/openfigi/client.py 
def __init__(self, api_key: str | None = None, timeout: float = 30.0) -> None:
    """init the HTTP client."""

    headers: dict[str, str] = {
        "Content-Type": "application/json"
    }
    self._api_key = api_key
    if api_key:
        headers["X-OPENFIGI-APIKEY"] = api_key
        self._limits = self._AUTHENTICATED_LIMITS
    else:
        self._limits = self._ANONYMOUS_LIMITS
    self._rate_limiter = RateLimiter(requests_per_minute=self._limits.requests_per_minute)
    self._client = httpx.Client(
        headers=headers,
        timeout=timeout,
    )

__enter__() -> Self

Source code in src/equities_classifier/clients/openfigi/client.py 
def __enter__(self) -> Self:
    return self

__exit__(*_: object) -> None

Source code in src/equities_classifier/clients/openfigi/client.py 
def __exit__(self, *_: object) -> None:
    self.close()

close() -> None

Close the HTTP client.

Source code in src/equities_classifier/clients/openfigi/client.py 
def close(self) -> None:
    """Close the HTTP client."""
    self._client.close()

read_provider_base_data(source_identifiers: Sequence[SecurityIdentifier], raise_error: bool = True) -> list[OpenFIGIRecord]

Read base date for one or more identifiers from OpenFIGI.

Source code in src/equities_classifier/clients/openfigi/client.py 
def read_provider_base_data(
    self,
    source_identifiers: Sequence[SecurityIdentifier],
    raise_error: bool = True
) -> list[OpenFIGIRecord]:
    """Read base date for one or more identifiers from OpenFIGI."""

    results: list[OpenFIGIRecord] = []

    batches = self._create_batches(source_identifiers)
    for batch in batches:

        response_data = self._execute_request(batch)
        for source_identifier, item in zip(batch, response_data, strict=True):
            # parsing accroSs batches for comprehension of data for duplicate source identifiers
            # (i.e. ticker and ISIN)
            self._parse_records(
                item=item,
                source_identifier=source_identifier,
                records=results,
                raise_error=raise_error
            )

    return results

remove_records_without_share_class_figi(records: list[OpenFIGIRecord]) -> list[OpenFIGIRecord] staticmethod

Remove records without share_class_FIGI if a ISIN-matching record exists.

Source code in src/equities_classifier/clients/openfigi/client.py 
@staticmethod
def remove_records_without_share_class_figi(
    records: list[OpenFIGIRecord],
) -> list[OpenFIGIRecord]:
    """Remove records without share_class_FIGI if a ISIN-matching record exists."""

    identifiers_with_share_class_figi: set[tuple[SecurityIdentifierType, str]] = {
        (identifier.type, str(identifier.value_cleaned))
        for record in records
        if record.share_class_figi is not None
        for identifier in record.identifiers
        if identifier.type == SecurityIdentifierType.ISIN
    }

    return [
        record
        for record in records
        if (
            record.share_class_figi is not None
            or not any(
                (identifier.type, identifier.value_cleaned)
                in identifiers_with_share_class_figi
                for identifier in record.identifiers
                if identifier.type == SecurityIdentifierType.ISIN
            )
        )
    ]

OpenFIGIResponseError

OpenFIGI returned an error response.