Client¶

class EventimClient:
    """Main client for querying the Eventim public and private APIs.

    Args:
        market: The Eventim market to query (e.g. ``EventimMarket.GERMANY``).
        proxies: Optional list of proxy URLs. When provided, the public fetcher picks
            one at random per request and the private fetcher cycles through them in
            order. Both fetchers use the proxy list by default; set
            ``use_public_proxy=False`` or ``use_private_proxy=False`` to opt a specific
            fetcher out at construction time.
        use_public_proxy: Whether the public (JSON API / httpx) fetcher should use
            the proxy list. Only meaningful when ``proxies`` is not ``None``.
            Defaults to ``True``.
        use_private_proxy: Whether the private (HTML scraping / Scrapling) fetcher
            should use the proxy list. Only meaningful when ``proxies`` is not ``None``.
            Defaults to ``True``.
        max_retries: Maximum number of retry attempts for public API requests on
            transient failures (HTTP 429, HTTP 5xx, connection resets, timeouts).
            Defaults to ``5``.
        retry_base_delay: Initial backoff delay in seconds before the first public API
            retry. Each subsequent retry doubles this delay. Full jitter is applied.
            Defaults to ``1.0``.
    """

    def __init__(
        self,
        market: EventimMarket,
        proxies: list[str] | None = None,
        use_public_proxy: bool = True,
        use_private_proxy: bool = True,
        max_retries: int = 5,
        retry_base_delay: float = 1.0,
    ) -> None:
        """Initialise the client for the given Eventim market.

        Args:
            market: The Eventim market to query (e.g. ``EventimMarket.GERMANY``).
            proxies: Optional list of proxy URLs. When provided, the public fetcher
                picks one at random per request and the private fetcher cycles through
                them in order. Pass ``None`` (default) to make all requests without a
                proxy.
            use_public_proxy: Whether the public (JSON API / httpx) fetcher should use
                the proxy list. Only meaningful when ``proxies`` is not ``None``.
                Defaults to ``True``.
            use_private_proxy: Whether the private (HTML scraping / Scrapling) fetcher
                should use the proxy list. Only meaningful when ``proxies`` is not
                ``None``. Defaults to ``True``.
            max_retries: Maximum number of retry attempts for public API requests on
                transient failures (HTTP 429, HTTP 5xx, connection resets, timeouts).
                Defaults to ``5``.
            retry_base_delay: Initial backoff delay in seconds before the first public
                API retry. Each subsequent retry doubles this delay. Full jitter is
                applied. Defaults to ``1.0``.
        """
        self.public_fetcher = EventimPublicFetcher(
            market.web_id,
            proxies=proxies if use_public_proxy else None,
            max_retries=max_retries,
            retry_base_delay=retry_base_delay,
        )

        self.private_fetcher: EventimPrivateFetcher = EventimPrivateFetcher(
            base_url=market.base_url,
            web_id=market.web_id,
            # ProxyRotator accepts list[str | dict[str, str]]; EventimClient exposes
            # list[str] since httpx (public fetcher) only supports string proxy URLs.
            # The cast is safe: every str satisfies str | dict[str, str].
            proxies=cast("list[str | dict[str, str]] | None", proxies if use_private_proxy else None),
        )

    def attractions(
        self,
        search_term: str,
        start_page: int | None = None,
        page_limit: int | None = None,
        raise_for_status: bool = False,
        use_proxy: bool = True,
    ) -> Iterator[EventimPublicAttraction]:
        """Search for attractions (artists, performers) by name.

        Args:
            search_term: The search query to filter attractions.
            start_page: Page number to start from, defaults to 1.
            page_limit: Maximum number of pages to fetch. Fetches all if not set.
            raise_for_status: If ``True``, raises on HTTP errors.
            use_proxy: If ``True`` (default) and proxies are configured on the client,
                they are used for this call. Set to ``False`` to bypass the proxy for
                this specific call.

        Returns:
            An iterator of EventimPublicAttraction objects.
        """
        for attraction_json in self.public_fetcher.paginate(
            endpoint="attractions",
            start_page=start_page or 1,
            page_limit=page_limit,
            extra_params={
                "search_term": search_term,
            },
            api_version="v1",
            raise_for_status=raise_for_status,
            use_proxy=use_proxy,
        ):
            yield EventimPublicAttraction(**attraction_json)

    def product_groups(
        self,
        categories: list[EventimCategory],
        sort: PUBLIC_SORT = "DateAsc",
        search_term: str | None = None,
        start_page: int | None = None,
        page_limit: int | None = None,
        page_size: int | None = None,
        raise_for_status: bool = False,
        use_proxy: bool = True,
    ) -> Iterator[EventimPublicProductGroup]:
        """Search for product groups (events/shows) filtered by category.

        Args:
            categories: List of ``EventimCategory`` members to filter by. Only the strings
                recognised by the current market are sent; if none of the requested categories
                exist for this market, an empty iterator is returned immediately.
            sort: Sort order for results, defaults to ``DateAsc``.
            search_term: Optional search query to narrow results.
            start_page: Page number to start from, defaults to 1.
            page_limit: Maximum number of pages to fetch. Fetches all if not set.
            page_size: Number of results per page. Uses the API default (20) when not set.
            raise_for_status: If ``True``, raises on HTTP errors.
            use_proxy: If ``True`` (default) and proxies are configured on the client,
                they are used for this call. Set to ``False`` to bypass the proxy for
                this specific call.

        Returns:
            An iterator of EventimPublicProductGroup objects.
        """
        web_id = self.public_fetcher.web_id
        categories_str = ",".join(s for c in categories for s in c.value.get(web_id, []))
        if not categories_str:
            return

        extra_params = {
            "sort": sort,
            "categories": categories_str,
        }

        if search_term:
            extra_params["search_term"] = search_term

        if page_size is not None:
            extra_params["top"] = str(page_size)

        for product_group_json in self.public_fetcher.paginate(
            endpoint="productGroups",
            start_page=start_page or 1,
            page_limit=page_limit,
            extra_params=extra_params,
            raise_for_status=raise_for_status,
            use_proxy=use_proxy,
        ):
            yield EventimPublicProductGroup(**product_group_json)

    def locations(
        self,
        search_term: str,
        start_page: int | None = None,
        page_limit: int | None = None,
        raise_for_status: bool = False,
        use_proxy: bool = True,
    ) -> Iterator[EventimPublicLocation]:
        """Search for venues and locations by name.

        Args:
            search_term: The search query to filter locations.
            start_page: Page number to start from, defaults to 1.
            page_limit: Maximum number of pages to fetch. Fetches all if not set.
            raise_for_status: If ``True``, raises on HTTP errors.
            use_proxy: If ``True`` (default) and proxies are configured on the client,
                they are used for this call. Set to ``False`` to bypass the proxy for
                this specific call.

        Returns:
            An iterator of EventimPublicLocation objects.
        """
        for locations_json in self.public_fetcher.paginate(
            endpoint="locations",
            start_page=start_page or 1,
            page_limit=page_limit,
            extra_params={
                "search_term": search_term,
            },
            api_version="v1",
            raise_for_status=raise_for_status,
            use_proxy=use_proxy,
        ):
            yield EventimPublicLocation(**locations_json)

    def content(
        self,
        search_term: str | None = None,
        start_page: int | None = None,
        page_limit: int | None = None,
        raise_for_status: bool = False,
        use_proxy: bool = True,
    ) -> Iterator[EventimPublicContent]:
        """Fetch editorial content items (e.g. FAQs).

        Args:
            search_term: Optional search query to filter content.
            start_page: Page number to start from, defaults to 1.
            page_limit: Maximum number of pages to fetch. Fetches all if not set.
            raise_for_status: If ``True``, raises on HTTP errors.
            use_proxy: If ``True`` (default) and proxies are configured on the client,
                they are used for this call. Set to ``False`` to bypass the proxy for
                this specific call.

        Returns:
            An iterator of EventimPublicContent objects.
        """
        for content_json in self.public_fetcher.paginate(
            endpoint="content",
            start_page=start_page or 1,
            page_limit=page_limit,
            extra_params={},
            api_version="v1",
            raise_for_status=raise_for_status,
            use_proxy=use_proxy,
        ):
            yield EventimPublicContent(**content_json)

    def products(
        self,
        product_group_id: int,
        start_page: int | None = None,
        page_limit: int | None = None,
        start_date: datetime | None = None,
        end_date: datetime | None = None,
        ticket_type: PRIVATE_TICKET_TYPE | None = None,
        city_name: str | None = None,
        raise_for_status: bool = False,
        use_proxy: bool = True,
    ) -> Iterator[EventimPrivateCalendarContentEvent]:
        """Fetch individual event dates (products) for a given product group.

        Args:
            product_group_id: The Eventim product group ID to fetch events for.
            start_page: Page number to start from, defaults to 1.
            page_limit: Maximum number of pages to fetch. Fetches all if not set.
            start_date: Filter events on or after this date.
            end_date: Filter events on or before this date.
            ticket_type: Filter by ticket type (``tickets`` or ``vip_tickets``).
            city_name: Filter events by city name.
            raise_for_status: If ``True``, raises on HTTP or parsing failures.
            use_proxy: If ``True`` (default) and proxies are configured on the client,
                they are used for this call. Set to ``False`` to bypass the proxy for
                this specific call.

        Returns:
            An iterator of EventimPrivateCalendarContentEvent objects.
        """
        extra_params: dict[str, str] = {}
        if start_date:
            extra_params["startdate"] = start_date.strftime("%Y-%m-%d")
        if end_date:
            extra_params["enddate"] = end_date.strftime("%Y-%m-%d")

        if ticket_type:
            extra_params["ptype"] = ticket_type

        if city_name:
            extra_params["cityname"] = city_name

        for product_json in self.private_fetcher.paginate_components(
            product_group_id,
            start_page=start_page or 1,
            page_limit=page_limit,
            extra_params=extra_params,
            raise_for_status=raise_for_status,
            use_proxy=use_proxy,
        ):
            yield EventimPrivateCalendarContentEvent(**product_json)