Client¶

class OxideHTTP:
    """Asynchronous HTTP client built on top of pyreqwest.

    The client provides support for standard HTTP requests, streamed
    responses, Server-Sent Events (SSE), middleware, proxies, configurable
    request settings, and optional response caching.

    The client is intended to be used as an asynchronous context manager::

        async with OxideHTTP() as client:
            response = await client.get('/users')

    It can also be explicitly closed using :meth:`close`.

    :param config: Client configuration. If omitted, an
        :class:`OxideHTTPConfig` with default values is used.
    :param cache: Optional response cache implementing
        :class:`CacheProtocol`.
    :param logger: Optional logger used for HTTP request, response, and SSE
        logging. Defaults to the ``oxidehttp`` logger.
    """

    __slots__ = ('_cache', '_client', '_config', '_logger')

    def __init__(
        self,
        config: OxideHTTPConfig | None = None,
        *,
        cache: CacheProtocol | None = None,
        logger: Logger | None = None,
    ) -> None:
        """Initialize an :class:`OxideHTTP` client.

        :param config: Client configuration or ``None`` to use the default
            configuration.
        :param cache: Optional cache implementation used for caching
            responses.
        :param logger: Optional logger used for HTTP and SSE diagnostics.
        """
        self._config = config or OxideHTTPConfig()
        self._client = self._build_client(self._config)
        self._cache = cache
        self._logger = logger or getLogger('oxidehttp')

    async def __aenter__(self) -> Self:
        """Enter the asynchronous context manager.

        :return: This client instance.
        :rtype: Self
        """
        await self._client.__aenter__()
        return self

    async def __aexit__(self, *args: object, **kwargs: Any) -> None:
        """Exit the asynchronous context manager.

        The underlying pyreqwest client is closed as part of exiting the
        context.

        :param args: Positional arguments supplied by the async context
            manager protocol.
        :param kwargs: Keyword arguments supplied by the async context manager
            protocol.
        """
        await self._client.__aexit__(*args, **kwargs)

    @staticmethod
    def _translate_exception(
        exc: pe.PyreqwestError | pe.DetailedPyreqwestError,
    ) -> e.OxideHTTPError:
        if isinstance(exc, pe.DetailedPyreqwestError):
            details = exc.details
        else:
            details = None

        for pyreqwest_type, oxide_type in _EXCEPTION_MAP.items():
            if isinstance(exc, pyreqwest_type):
                return oxide_type(str(exc), original=exc, details=details)

        return e.OxideHTTPError(str(exc), original=exc, details=details)

    @staticmethod
    def _build_client(config: OxideHTTPConfig) -> Client:
        """Build a pyreqwest client from the supplied configuration.

        The configuration controls runtime behavior, connection limits,
        TLS certificate verification, base URL, default headers, user agent,
        timeout, proxies, and client-level middleware.

        :param config: Configuration used to initialize the pyreqwest client.
        :return: A configured pyreqwest client.
        :rtype: Client
        """
        client = ClientBuilder()
        proxy_builders = {
            'http': ProxyBuilder.http,
            'https': ProxyBuilder.https,
            'all': ProxyBuilder.all,
        }

        connection = config.connection
        security = config.security
        request = config.request
        compress = config.compression

        # Base config
        if config.base_url is not None:
            client.base_url(config.base_url.rstrip('/') + '/')

        if config.headers is not None:
            client.default_headers(config.headers)

        # Connection config
        client.runtime_multithreaded(enable=connection.runtime_multithreaded)

        if connection.max_connections is not None:
            client.max_connections(connection.max_connections)

        if connection.timeout is not None:
            client.timeout(connection.timeout)

        for proxy_type, proxy_url in connection.proxies.items():
            proxy = proxy_builders[proxy_type](proxy_url)
            client.proxy(proxy)

        # Security config
        client.danger_accept_invalid_certs(enable=not security.verify_ssl)

        # Request config
        if request.user_agent is not None:
            client.user_agent(request.user_agent)

        for middleware in request.middlewares:
            client.with_middleware(middleware)

        # Compression config
        client.gzip(compress.gzip)
        client.brotli(compress.brotli)
        client.zstd(compress.zstd)
        client.deflate(compress.deflate)

        return client.build()

    @staticmethod
    def clean_dict[K, V](params: Mapping[K, V | None]) -> dict[K, V]:
        """Remove entries with ``None`` values from a mapping.

        A new dictionary is returned; the input mapping is not modified.

        :param params: Mapping containing values that may be ``None``.
        :return: Dictionary containing only entries whose values are not
            ``None``.
        :rtype: dict[K, V]
        """
        return {k: v for k, v in params.items() if v is not None}

    @staticmethod
    def _add_params(builder: RequestBuilder, params: Params) -> RequestBuilder:
        """Add query parameters to a request builder.

        Sequence values are emitted as repeated query parameters. Boolean
        values are serialized as lowercase ``true`` or ``false`` strings.
        Other values are converted to strings.

        For example::

            {'tag': ['python', 'http']}

        becomes two query parameters equivalent to::

            ?tag=python&tag=http

        :param builder: Request builder to which the parameters are added.
        :param params: Query parameters to add.
        :return: The same request builder with the query parameters applied.
        :rtype: RequestBuilder
        """
        for key, value in params.items():
            if isinstance(value, Sequence) and not isinstance(
                value, (str, bytes, bytearray)
            ):
                for item in value:
                    builder.query({key: str(item)})

            elif isinstance(value, bool):
                builder.query({key: 'true' if value else 'false'})

            else:
                builder.query({key: str(value)})

        return builder

    def _get_method(self, method: HTTPMethod) -> RequestMethod:
        """Resolve an HTTP method to the corresponding client method.

        :param method: HTTP method to resolve.
        :return: Bound request method from the underlying client.
        :rtype: RequestMethod
        :raises KeyError: If ``method`` is not supported by the client.
        """
        return {
            'GET': self._client.get,
            'POST': self._client.post,
            'PUT': self._client.put,
            'PATCH': self._client.patch,
            'DELETE': self._client.delete,
            'HEAD': self._client.head,
        }[method]

    @staticmethod
    def _set_body(
        builder: RequestBuilder,
        *,
        json: JSON | None = None,
        body_text: str | None = None,
        body_bytes: bytes | None = None,
        form: Form | None = None,
    ) -> None:
        """Apply the first non-None body representation."""
        if json is not None:
            builder.body_json(json)
        elif body_text is not None:
            builder.body_text(body_text)
        elif body_bytes is not None:
            builder.body_bytes(body_bytes)
        elif form is not None:
            builder.form(form)

    def _build_request(
        self,
        method: HTTPMethod,
        url: str,
        params: Params | None = None,
        headers: Headers | None = None,
        json: JSON | None = None,
        body_text: str | None = None,
        body_bytes: bytes | None = None,
        form: Form | None = None,
        middlewares: Sequence[Middleware] | None = None,
        extensions: Extensions | None = None,
    ) -> RequestBuilder:
        """Build a configured HTTP request.

        Relative URLs beginning with ``/`` have the leading slash removed
        before being passed to the underlying client. This allows them to be
        resolved against a configured base URL.

        Request body representations are applied in the following order:

        1. ``json``
        2. ``body_text``
        3. ``body_bytes``
        4. ``form``

        Therefore, when multiple body arguments are provided, the first
        non-``None`` value in that order takes precedence.

        :param method: HTTP method to use.
        :param url: Request URL or path.
        :param params: Optional query parameters.
        :param headers: Optional request headers.
        :param json: Optional JSON request body.
        :param body_text: Optional text request body.
        :param body_bytes: Optional raw binary request body.
        :param form: Optional form-encoded request body.
        :param middlewares: Optional request-specific middleware.
        :param extensions: Optional request extensions.
        :return: Configured request builder.
        :rtype: RequestBuilder
        """
        url = url.removeprefix('/')

        try:
            builder = self._get_method(method)(url)

            if params is not None:
                self._add_params(builder, params)

            self._set_body(
                builder,
                json=json,
                body_text=body_text,
                body_bytes=body_bytes,
                form=form,
            )

            if headers is not None:
                builder.headers(headers)

            if middlewares is not None:
                for middleware in middlewares:
                    builder.with_middleware(middleware)

            if extensions is not None:
                builder.extensions(extensions)

        except pe.PyreqwestError as exc:
            raise self._translate_exception(exc) from exc

        return builder

    async def _send_request(
        self,
        method: HTTPMethod,
        url: str,
        *,
        params: Params | None = None,
        headers: Headers | None = None,
        json: JSON | None = None,
        body_text: str | None = None,
        body_bytes: bytes | None = None,
        form: Form | None = None,
        middlewares: Sequence[Middleware] | None = None,
        extensions: Extensions | None = None,
    ) -> PyreqwestResponse:
        """Build and send an HTTP request.

        This method returns the raw response from pyreqwest. Higher-level
        methods convert it into the package's response type.

        :param method: HTTP method to use.
        :param url: Request URL or path.
        :param params: Optional query parameters.
        :param headers: Optional request headers.
        :param json: Optional JSON request body.
        :param body_text: Optional text request body.
        :param body_bytes: Optional raw binary request body.
        :param form: Optional form-encoded request body.
        :param middlewares: Optional request-specific middleware.
        :param extensions: Optional request extensions.
        :return: Raw pyreqwest response.
        :rtype: PyreqwestResponse
        :raises PyreqwestError: If the underlying request fails.
        """
        builder = self._build_request(
            method,
            url,
            params=params,
            headers=headers,
            json=json,
            body_text=body_text,
            body_bytes=body_bytes,
            form=form,
            middlewares=middlewares,
            extensions=extensions,
        )

        try:
            req = builder.build()
            return await req.send()

        except pe.PyreqwestError as exc:
            raise self._translate_exception(exc) from exc

    async def _request(
        self,
        method: HTTPMethod,
        url: str,
        *,
        params: Params | None = None,
        headers: Headers | None = None,
        json: JSON | None = None,
        body_text: str | None = None,
        body_bytes: bytes | None = None,
        form: Form | None = None,
        middlewares: Sequence[Middleware] | None = None,
        extensions: Extensions | None = None,
        cache_key: str | None = None,
        cache_ttl: float | None = None,
    ) -> s.BaseResponse:
        """Execute an HTTP request and optionally cache its response.

        The cache key is derived from the request URL, query parameters, and
        explicit ``cache_key`` value. Request headers and body contents are
        not included in the generated cache key.

        If ``cache_ttl`` is supplied without a configured cache, the request
        proceeds normally and a warning is logged.

        :param method: HTTP method to use.
        :param url: Request URL or path.
        :param params: Optional query parameters.
        :param headers: Optional request headers.
        :param json: Optional JSON request body.
        :param body_text: Optional text request body.
        :param body_bytes: Optional raw binary request body.
        :param form: Optional form-encoded request body.
        :param middlewares: Optional request-specific middleware.
        :param extensions: Optional request extensions.
        :param cache_key: Optional additional value used to distinguish cache
            entries.
        :param cache_ttl: Optional response cache lifetime in seconds.
        :return: Package-level HTTP response.
        :rtype: s.BaseResponse
        :raises PyreqwestError: If the underlying request fails.
        """
        request_headers = dict(headers or {})
        cache_key_encoded = sha256(
            encode({'url': url, 'params': params, 'cache_key': cache_key})
        ).hexdigest()

        if cache_ttl is not None and cache_ttl > 0:
            if self._cache is None:
                self._logger.warning(
                    'cached is not initialized inside OxideHTTP client.'
                )

            else:
                cached = await self._cache.get(cache_key_encoded)

                if cached is not None:
                    return cached

        self._logger.debug(
            'HTTP request', extra={'method': method, 'url': url}
        )

        response = await self._send_request(
            method,
            url,
            params=params,
            headers=request_headers,
            json=json,
            body_text=body_text,
            body_bytes=body_bytes,
            form=form,
            middlewares=middlewares,
            extensions=extensions,
        )

        self._logger.debug(
            'HTTP response',
            extra={'method': method, 'url': url, 'status': response.status},
        )

        response = s.Response(pyreqwest_response=response)

        if (
            cache_key is not None
            and self._cache is not None
            and cache_ttl is not None
            and cache_ttl > 0
        ):
            await self._cache.set(
                cache_key_encoded,
                status_code=response.status_code,
                headers=response.headers,
                data=(await response.get_bytes()),
                ttl=cache_ttl,
            )

        return response

    @asynccontextmanager
    async def _stream_request(
        self,
        method: HTTPMethod,
        url: str,
        *,
        params: Params | None = None,
        headers: Headers | None = None,
        json: JSON | None = None,
        body_text: str | None = None,
        body_bytes: bytes | None = None,
        form: Form | None = None,
        buffer_limit: int = 100,
        middlewares: Sequence[Middleware] | None = None,
        extensions: Extensions | None = None,
    ) -> AsyncGenerator[AsyncIterator[bytes]]:
        """Stream an HTTP response body.

        The underlying response is kept open for the lifetime of the async
        generator. Response data is read in 16-byte chunks and yielded as
        ``bytes``.

        :param method: HTTP method to use.
        :param url: Request URL or path.
        :param params: Optional query parameters.
        :param headers: Optional request headers.
        :param json: Optional JSON request body.
        :param body_text: Optional text request body.
        :param body_bytes: Optional raw binary request body.
        :param form: Optional form-encoded request body.
        :param buffer_limit: Maximum size of the underlying streamed-read
            buffer.
        :param middlewares: Optional request-specific middleware.
        :param extensions: Optional request extensions.
        :yield: Raw response body chunks.
        :ytype: bytes
        :raises PyreqwestError: If the underlying request fails.
        :raises asyncio.CancelledError: If the operation is cancelled.
        """
        builder = self._build_request(
            method,
            url,
            params=params,
            headers=headers,
            json=json,
            body_text=body_text,
            body_bytes=body_bytes,
            form=form,
            middlewares=middlewares,
            extensions=extensions,
        )
        builder.streamed_read_buffer_limit(buffer_limit)

        async with builder.build_streamed() as response:

            async def iter_chunks() -> AsyncIterator[bytes]:
                try:
                    while (
                        chunk := await response.body_reader.read_chunk()
                    ) is not None:
                        yield bytes(chunk)
                except CancelledError:
                    raise
                except pe.PyreqwestError as exc:
                    raise self._translate_exception(exc) from exc

            yield iter_chunks()

    async def close(self) -> None:
        """Close the underlying HTTP client.

        After this method completes, the client should no longer be used for
        making requests.

        :raises PyreqwestError: If closing the underlying client fails.
        """
        await self._client.close()

    async def get(
        self,
        url: str,
        params: Params | None = None,
        headers: Headers | None = None,
        middlewares: Sequence[Middleware] | None = None,
        extensions: Extensions | None = None,
        cache_ttl: float | None = None,
    ) -> s.BaseResponse:
        """Send an HTTP GET request.

        :param url: Request URL or path. Relative paths are resolved against
            the configured base URL.
        :param params: Optional query parameters.
        :param headers: Optional request headers.
        :param middlewares: Optional request-specific middleware.
        :param extensions: Optional request extensions.
        :param cache_ttl: Optional response cache lifetime in seconds.
        :return: HTTP response.
        :rtype: s.BaseResponse
        :raises PyreqwestError: If the request fails.
        """
        return await self._request(
            'GET',
            url,
            params=params,
            headers=headers,
            middlewares=middlewares,
            extensions=extensions,
            cache_ttl=cache_ttl,
        )

    async def post(
        self,
        url: str,
        headers: dict[Any, Any] | None = None,
        params: dict[Any, Any] | None = None,
        json: dict[Any, Any] | None = None,
        body_text: str | None = None,
        body_bytes: bytes | None = None,
        form: dict[Any, Any] | None = None,
        middlewares: Sequence[Middleware] | None = None,
        extensions: Extensions | None = None,
    ) -> s.BaseResponse:
        """Send an HTTP POST request.

        :param url: Request URL or path.
        :param headers: Optional request headers.
        :param params: Optional query parameters.
        :param json: Optional JSON request body.
        :param body_text: Optional text request body.
        :param body_bytes: Optional raw binary request body.
        :param form: Optional form-encoded request body.
        :param middlewares: Optional request-specific middleware.
        :param extensions: Optional request extensions.
        :return: HTTP response.
        :rtype: s.BaseResponse
        :raises PyreqwestError: If the request fails.
        """
        return await self._request(
            'POST',
            url,
            params=params,
            headers=headers,
            json=json,
            body_text=body_text,
            body_bytes=body_bytes,
            form=form,
            middlewares=middlewares,
            extensions=extensions,
        )

    async def put(
        self,
        url: str,
        headers: dict[Any, Any] | None = None,
        params: dict[Any, Any] | None = None,
        json: dict[Any, Any] | None = None,
        body_text: str | None = None,
        body_bytes: bytes | None = None,
        middlewares: Sequence[Middleware] | None = None,
        extensions: Extensions | None = None,
    ) -> s.BaseResponse:
        """Send an HTTP PUT request.

        :param url: Request URL or path.
        :param headers: Optional request headers.
        :param params: Optional query parameters.
        :param json: Optional JSON request body.
        :param body_text: Optional text request body.
        :param body_bytes: Optional raw binary request body.
        :param middlewares: Optional request-specific middleware.
        :param extensions: Optional request extensions.
        :return: HTTP response.
        :rtype: s.BaseResponse
        :raises PyreqwestError: If the request fails.
        """
        return await self._request(
            'PUT',
            url,
            params=params,
            headers=headers,
            json=json,
            body_text=body_text,
            body_bytes=body_bytes,
            middlewares=middlewares,
            extensions=extensions,
        )

    async def patch(
        self,
        url: str,
        headers: dict[Any, Any] | None = None,
        params: dict[Any, Any] | None = None,
        json: dict[Any, Any] | None = None,
        body_text: str | None = None,
        body_bytes: bytes | None = None,
        middlewares: Sequence[Middleware] | None = None,
        extensions: Extensions | None = None,
    ) -> s.BaseResponse:
        """Send an HTTP PATCH request.

        :param url: Request URL or path.
        :param headers: Optional request headers.
        :param params: Optional query parameters.
        :param json: Optional JSON request body.
        :param body_text: Optional text request body.
        :param body_bytes: Optional raw binary request body.
        :param middlewares: Optional request-specific middleware.
        :param extensions: Optional request extensions.
        :return: HTTP response.
        :rtype: s.BaseResponse
        :raises PyreqwestError: If the request fails.
        """
        return await self._request(
            'PATCH',
            url,
            params=params,
            headers=headers,
            json=json,
            body_text=body_text,
            body_bytes=body_bytes,
            middlewares=middlewares,
            extensions=extensions,
        )

    async def delete(
        self,
        url: str,
        headers: dict[Any, Any] | None = None,
        params: dict[Any, Any] | None = None,
        json: dict[Any, Any] | None = None,
        middlewares: Sequence[Middleware] | None = None,
        extensions: Extensions | None = None,
    ) -> s.BaseResponse:
        """Send an HTTP DELETE request.

        :param url: Request URL or path.
        :param headers: Optional request headers.
        :param params: Optional query parameters.
        :param json: Optional JSON request body.
        :param middlewares: Optional request-specific middleware.
        :param extensions: Optional request extensions.
        :return: HTTP response.
        :rtype: s.BaseResponse
        :raises PyreqwestError: If the request fails.
        """
        return await self._request(
            'DELETE',
            url,
            params=params,
            headers=headers,
            json=json,
            middlewares=middlewares,
            extensions=extensions,
        )

    @asynccontextmanager
    async def stream(
        self,
        method: HTTPMethod,
        url: str,
        *,
        params: Params | None = None,
        headers: Headers | None = None,
        json: JSON | None = None,
        body_text: str | None = None,
        body_bytes: bytes | None = None,
        form: Form | None = None,
        buffer_limit: int = 100,
        middlewares: Sequence[Middleware] | None = None,
        extensions: Extensions | None = None,
    ) -> AsyncGenerator[bytes]:
        """Stream an HTTP response body.

        The response is exposed as an async generator of raw byte chunks.

        :param method: HTTP method to use.
        :param url: Request URL or path.
        :param params: Optional query parameters.
        :param headers: Optional request headers.
        :param json: Optional JSON request body.
        :param body_text: Optional text request body.
        :param body_bytes: Optional raw binary request body.
        :param form: Optional form-encoded request body.
        :param buffer_limit: Maximum size of the underlying streamed-read
            buffer.
        :param middlewares: Optional request-specific middleware.
        :param extensions: Optional request extensions.
        :yield: Raw response body chunks.
        :ytype: bytes
        :raises PyreqwestError: If the underlying request fails.
        :raises asyncio.CancelledError: If the stream is cancelled.
        """
        async with self._stream_request(
            method,
            url,
            params=params,
            headers=headers,
            json=json,
            body_text=body_text,
            body_bytes=body_bytes,
            form=form,
            buffer_limit=buffer_limit,
            middlewares=middlewares,
            extensions=extensions,
        ) as chunks:
            async for chunk in chunks:
                yield chunk

    async def _iter_sse_events(
        self,
        method: HTTPMethod,
        url: str,
        *,
        params: Params | None = None,
        headers: Headers,
        json: JSON | None = None,
        body_text: str | None = None,
        body_bytes: bytes | None = None,
        form: Form | None = None,
        buffer_limit: int = 0,
        middlewares: Sequence[Middleware] | None = None,
        extensions: Extensions | None = None,
    ) -> AsyncGenerator[SSEEvent]:
        """Consume and parse a single SSE connection."""
        parser = SSEParser()

        async with self._stream_request(
            method,
            url,
            params=params,
            headers=headers,
            json=json,
            body_text=body_text,
            body_bytes=body_bytes,
            form=form,
            buffer_limit=buffer_limit,
            middlewares=middlewares,
            extensions=extensions,
        ) as chunks:
            async for chunk in chunks:
                for event in parser.feed(chunk):
                    yield event

        for event in parser.finish():
            yield event

    def _handle_sse_event(
        self,
        event: SSEEvent,
        *,
        url: str,
        reconnects: int,
        last_event_id: str | None,
        server_retry: float | None,
    ) -> tuple[str | None, float | None]:
        """Update reconnect state and log a received event."""
        if event.event_id is not None:
            last_event_id = event.event_id

        if event.retry is not None:
            server_retry = event.retry

        self._logger.info(
            'SSE event',
            extra={'url': url, 'reconnect': reconnects, 'event': event},
        )

        return last_event_id, server_retry

    async def sse(
        self,
        method: HTTPMethod,
        url: str,
        *,
        params: Params | None = None,
        headers: Headers | None = None,
        json: JSON | None = None,
        body_text: str | None = None,
        body_bytes: bytes | None = None,
        form: Form | None = None,
        buffer_limit: int = 0,
        auto_reconnect: bool = False,
        reconnect_delay: float = 3.0,
        max_reconnects: int | None = None,
        middlewares: Sequence[Middleware] | None = None,
        extensions: Extensions | None = None,
    ) -> AsyncGenerator[SSEEvent]:
        """Consume a Server-Sent Events stream.

        Events are parsed incrementally from the response body and yielded as
        :class:`SSEEvent` instances.

        When ``auto_reconnect`` is enabled, a failed connection is
        re-established automatically. The most recently received event ID is
        sent using the ``Last-Event-ID`` header.

        If the server sends a ``retry`` value, that value is used as the
        reconnect delay instead of ``reconnect_delay``.

        The client automatically sets the following headers when they are not
        already present:

        * ``Accept: text/event-stream``
        * ``Cache-Control: no-cache``

        :param method: HTTP method used to establish the SSE connection.
        :param url: Request URL or path.
        :param params: Optional query parameters.
        :param headers: Optional request headers.
        :param json: Optional JSON request body.
        :param body_text: Optional text request body.
        :param body_bytes: Optional raw binary request body.
        :param form: Optional form-encoded request body.
        :param buffer_limit: Maximum size of the underlying streamed-read
            buffer.
        :param auto_reconnect: Whether to automatically reconnect after a
            :class:`PyreqwestError`.
        :param reconnect_delay: Default delay between reconnect attempts,
            in seconds.
        :param max_reconnects: Maximum number of reconnect attempts.
            ``None`` allows unlimited attempts.
        :param middlewares: Optional request-specific middleware.
        :param extensions: Optional request extensions.
        :yield: Parsed Server-Sent Events.
        :ytype: SSEEvent
        :raises ValueError: If ``reconnect_delay`` is negative.
        :raises ValueError: If ``max_reconnects`` is negative.
        :raises PyreqwestError: If a connection error occurs while
            ``auto_reconnect`` is disabled.
        :raises asyncio.CancelledError: If the stream is cancelled.
        """
        if reconnect_delay < 0:
            err = 'reconnect_delay cannot be negative'
            raise ValueError(err)

        if max_reconnects is not None and max_reconnects < 0:
            err = 'max_reconnects cannot be negative'
            raise ValueError(err)

        reconnects = 0
        last_event_id: str | None = None
        server_retry: float | None = None

        request_headers = dict(headers or {})
        request_headers.setdefault('Accept', 'text/event-stream')
        request_headers.setdefault('Cache-Control', 'no-cache')

        self._logger.info(
            'SSE connect', extra={'url': url, 'reconnect': reconnects}
        )

        while True:
            if last_event_id is not None:
                request_headers['Last-Event-ID'] = last_event_id
                self._logger.info(
                    'SSE reconnect',
                    extra={'url': url, 'reconnect': reconnects},
                )

            try:
                async for event in self._iter_sse_events(
                    method,
                    url,
                    params=params,
                    headers=request_headers,
                    json=json,
                    body_text=body_text,
                    body_bytes=body_bytes,
                    form=form,
                    buffer_limit=buffer_limit,
                    middlewares=middlewares,
                    extensions=extensions,
                ):
                    last_event_id, server_retry = self._handle_sse_event(
                        event,
                        url=url,
                        reconnects=reconnects,
                        last_event_id=last_event_id,
                        server_retry=server_retry,
                    )
                    yield event

            except CancelledError:
                raise

            except pe.PyreqwestError as exc:
                if not auto_reconnect:
                    raise self._translate_exception(exc) from exc

            else:
                if not auto_reconnect:
                    return

            if max_reconnects is not None and reconnects > max_reconnects:
                return

            reconnects += 1
            delay = (
                server_retry if server_retry is not None else reconnect_delay
            )

            if delay > 0:
                await sleep(delay)