Responses¶

class BaseResponse(ABC):
    is_cached: bool = False

    @property
    @abstractmethod
    def status_code(self) -> HTTPStatus: ...

    @property
    @abstractmethod
    def headers(self) -> Mapping[str, str]: ...

    @abstractmethod
    async def get_bytes(self) -> memoryview: ...

    async def text(self) -> str:
        data = await self.get_bytes()
        return str(data, encoding='utf-8', errors='replace')

    async def json(self) -> dict[Any, Any]:
        data = await self.get_bytes()

        try:
            return decode(data)

        except DecodeError as exc:
            raise e.OxideJSONDecodeError(str(exc), original=exc) from exc

@dataclass(slots=True)
class Response(BaseResponse):
    pyreqwest_response: PyreqwestResponse
    _bytes: bytes | None = None

    @property
    @override
    def status_code(self) -> HTTPStatus:
        return HTTPStatus(self.pyreqwest_response.status)

    @property
    @override
    def headers(self) -> Mapping[str, str]:
        return self.pyreqwest_response.headers

    @override
    async def get_bytes(self) -> memoryview:
        if self._bytes is None:
            self._bytes = bytes(await self.pyreqwest_response.bytes())

        return memoryview(self._bytes)

@dataclass(slots=True)
class CachedResponse(BaseResponse):
    _status_code: HTTPStatus
    _headers: Headers
    data: bytes
    is_cached: bool = True

    @property
    @override
    def status_code(self) -> HTTPStatus:
        return self._status_code

    @property
    @override
    def headers(self) -> Headers:
        return self._headers

    @override
    async def get_bytes(self) -> memoryview:
        return memoryview(self.data)

@dataclass(slots=True, frozen=True)
class SSEEvent:
    data: str
    event: str | None = None
    event_id: str | None = None
    retry: int | None = None