Class LaunchDarklyClient

api/integrations/launch_darkly/client.py:90–268

class LaunchDarklyClient:
    def __init__(self, token: str) -> None:
        client_session = Session()
        client_session.headers.update(
            {
                "Authorization": token,
                "LD-API-Version": LAUNCH_DARKLY_API_VERSION,
            }
        )
        self.client_session = client_session

    @launch_darkly_backoff
    def _get_json_response(
        self,
        endpoint: str,
        params: Optional[dict[str, Any]] = None,
    ) -> T:  # type: ignore[type-var]
        full_url = f"{LAUNCH_DARKLY_API_BASE_URL}{endpoint}"
        response = self.client_session.get(full_url, params=params)
        response.raise_for_status()
        return response.json()  # type: ignore[no-any-return]

    def _iter_paginated_items(
        self,
        collection_endpoint: str,
        additional_params: Optional[dict[str, Any]] = None,
        use_legacy_offset_pagination: bool = False,
    ) -> Iterator[T]:
        """
        Iterator over paginated items in the given collection endpoint.

        :param collection_endpoint: endpoint to get the collection of items
        :param additional_params: Additional parameters to include in the request
        :param use_legacy_offset_pagination: Whether to use offset based pagination if `next` links do not
        exist in the response. Some endpoints do not have `next` links and require offset based pagination.
        :return: Iterator over the items in the collection
        """
        params = {"limit": LAUNCH_DARKLY_API_ITEM_COUNT_LIMIT_PER_PAGE}
        offset = 0
        if additional_params:
            params.update(additional_params)

        response_json: dict[str, Any] = self._get_json_response(
            endpoint=collection_endpoint,
            params=params,
        )
        while True:
            items = response_json.get("items") or []
            yield from items
            links: Optional[dict[str, ld_types.Link]] = response_json.get("_links")
            if (
                links
                and (next_link := links.get("next"))
                and (next_endpoint := next_link.get("href"))
            ):
                # Don't specify params here because links.next.href includes the
                # original limit and calculates offsets accordingly.
                response_json = self._get_json_response(