Skip to content

Client

client

Client for Morningstar.

MorningstarClient(timeout: float = 30.0, seleniumwrapper: object | None = None, clean_nonUS_ticker: bool = True, test_wo_browser: bool = False)

Morningstar HTTP / API client.

Source code in src/equities_classifier/clients/morningstar/client.py 
def __init__(
    self,
    timeout: float = 30.0,
    seleniumwrapper: object | None = None,
    clean_nonUS_ticker: bool = True,
    test_wo_browser: bool = False,
) -> None:
    """Initialize Morningstar client."""

    if not test_wo_browser:
        if seleniumwrapper is None:
            options = uc.ChromeOptions()
            # options.add_argument("--headless=new")   # does not work with Morningstar because CloudFront-detected
            options.add_argument("--no-sandbox")
            options.add_argument("--disable-dev-shm-usage")
            options.add_argument("--disable-gpu")
            options.add_argument("--window-size=1920,1080")
            try:
                self._client = uc.Chrome(options=options)
            except Exception as e:
                print("Failed to start Chrome")
                raise e
            try:
                config = StabilizationConfig(
                    timeout=10,  # Max wait time (seconds)
                    network_idle_threshold=5,  # Max pending requests (allows background traffic)
                    strictness='normal',  # 'strict' | 'normal' | 'relaxed'
                    debug_mode=False  # Enable logging
                )
                self._client = stabilize(self._client, config=config)
            except StabilizationTimeout as e:
                print("Failed to stabilize Chrome  with 'waitless'")
                diagnostics = get_diagnostics(self._client)
                print_report(diagnostics)  # Print detailed report
                raise e
        else:
            self._client = seleniumwrapper
    else:
        self._client = None

    self._timeout = timeout

    self._access_token: str | None = None
    self._access_token_expires: datetime.date | None = None

    self._clean_nonUS_ticker = clean_nonUS_ticker

__enter__() -> Self

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

__exit__(*_: object) -> None

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

close() -> None

Release browser resources.

Source code in src/equities_classifier/clients/morningstar/client.py 
def close(self) -> None:
    """Release browser resources."""

    # check self_client before execution due to potential double-close when using pytest
    if self._client:
        self._client.close()
        self._client = None

leaf_paths(data: Mapping[str, Any], path: tuple[str, ...] = (), exclude_leaves: Collection[str] = []) -> list[tuple[str, ...]] staticmethod

Return all key paths from the root to every leaf.

Source code in src/equities_classifier/clients/morningstar/client.py 
@staticmethod
def leaf_paths(
    data: Mapping[str, Any],
    path: tuple[str, ...] = (),
    exclude_leaves: Collection[str] = []
) -> list[tuple[str, ...]]:
    """Return all key paths from the root to every leaf."""

    result: list[tuple[str, ...]] = []

    for key, value in data.items():

        if key in exclude_leaves:
            continue

        current = (*path, key)
        if isinstance(value, Mapping):
            result.extend(MorningstarClient.leaf_paths(value, current, exclude_leaves))
        else:
            result.append(current)

    return result

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

Read base data for one or more identifiers from Morningstar.

Source code in src/equities_classifier/clients/morningstar/client.py 
def read_provider_base_data(
    self,
    source_identifiers: Sequence[SecurityIdentifier],
    raise_error: bool = False
) -> list[MorningstarRecord]:
    """Read base data for one or more identifiers from Morningstar."""

    records: list[MorningstarRecord] = []

    for source_identifier in source_identifiers:

        if source_identifier.type not in self._MORNINGSTAR_IDENTIFIER_TYPES.values():
            ClientHelper.invalid_security_type(
                DataSourceID.MORNINGSTAR,
                source_identifier
            )
            continue

        response_data = self._execute_search_request(source_identifier)
        search_results = self._parse_search_results(
            source_identifier,
            response_data,
            raise_error,
        )
        search_results = [
            result for result in search_results
            if (
                   source_identifier.type == SecurityIdentifierType.ISIN
                   and result.isin == source_identifier.value
               )
               or (
                   source_identifier.type == SecurityIdentifierType.TICKER
                   and result.ticker == source_identifier.value_cleaned
               )
        ]

        if search_results is not None:
            self._parse_records(
                source_identifier,
                search_results,
                records,
                raise_error
            )

    return records

read_provider_profile_data(records: list[MorningstarRecord], raise_error: bool = False) -> list[MorningstarRecord]

Read profile data from Morningstar.

Source code in src/equities_classifier/clients/morningstar/client.py 
def read_provider_profile_data(
    self,
    records: list[MorningstarRecord],
    raise_error: bool = False
) -> list[MorningstarRecord]:
    """Read profile data from Morningstar."""

    for record in records:
        profile_data = self._execute_profile_request(
            security_id=record.company_id if record.company_id else record.performance_id[0]
        )
        record = self._parse_profile_to_record(profile_data, record, raise_error)

    return records

supports_identifier_type(identifier_type: SecurityIdentifierType) -> bool classmethod

Check if identifier type supported

Source code in src/equities_classifier/clients/morningstar/client.py 
@classmethod
def supports_identifier_type(cls, identifier_type: SecurityIdentifierType, ) -> bool:
    """Check if identifier type supported"""
    return identifier_type in cls._MORNINGSTAR_IDENTIFIER_TYPES

MorningstarResponseError

Morningstar returned an error response.