Skip to content

ProxyResponse

Proxy response object.

This object encapsulates the incoming request and the response from target server. It exposes a method to set the response object which can then be modified before being returned to the client.

Parameters:

Name Type Description Default
in_resp ClientResponse

The incoming response object.

 required
Source code in aiorp/response.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
class ProxyResponse:
    """Proxy response object.

    This object encapsulates the incoming request and the response from target server.
    It exposes a method to set the response object which can then be modified before being
    returned to the client.

    Args:
        in_resp: The incoming response object.
    """

    def __init__(
        self,
        in_resp: client.ClientResponse,
    ):
        """Initialize the proxy response object.

        Args:
            in_resp: The incoming response object.
        """
        self.in_resp: client.ClientResponse = in_resp
        self._web: web.StreamResponse | None = None
        self._content: bytes | None = None

    @property
    def web_response_set(self) -> bool:
        """Checks if the web response is set already.

        Returns:
            A boolean, true if set, false otherwise
        """
        return self._web is not None

    @property
    def web(
        self,
    ) -> StreamResponse | Response:
        """Access the web response.

        Returns:
            A response, either StreamResponse or Response.

        Raises:
            ValueError: When response is not set yet.
        """
        if self._web is None:
            raise ValueError("Response has not been set")
        return self._web

    async def set_response(
        self, response_type: ResponseType = ResponseType.BASE
    ) -> StreamResponse | Response:
        """Set the response using the given response type.

        Args:
            response_type: The type of response to set.

        Returns:
            The set web response.

        Raises:
            ValueError: When attempted to set the response a second time.
        """
        if self._web is not None:
            raise ValueError("Response can only be set once")
        if response_type == ResponseType.STREAM:
            self._web = await self._get_stream_response()
        else:
            self._web = await self._get_base_response()
        return self._web

    async def _get_stream_response(self) -> StreamResponse:
        """Convert incoming response to stream response."""

        headers = CIMultiDict(self.in_resp.headers)

        # These headers should not be proxied
        headers.pop("content-length", None)
        headers.pop("content-encoding", None)
        headers.pop("content-type", None)

        stream_resp = StreamResponse(
            status=self.in_resp.status,
            reason=self.in_resp.reason,
            headers=self.in_resp.headers,
        )
        return stream_resp

    async def _get_base_response(self) -> Response:
        """Convert incoming response to base response."""
        content = await self.in_resp.read()

        headers = CIMultiDict(self.in_resp.headers)

        # These headers should not be proxied
        headers.pop("content-length", None)
        headers.pop("content-encoding", None)

        if content:
            headers["content-length"] = str(len(content))

        resp = Response(
            status=self.in_resp.status,
            reason=self.in_resp.reason,
            headers=headers,
            body=content,
        )
        return resp

web property

Access the web response.

Returns:

Type Description
StreamResponse | Response

A response, either StreamResponse or Response.

Raises:

Type Description
ValueError

When response is not set yet.

web_response_set property

Checks if the web response is set already.

Returns:

Type Description
bool

A boolean, true if set, false otherwise

__init__(in_resp)

Initialize the proxy response object.

Parameters:

Name Type Description Default
in_resp ClientResponse

The incoming response object.

 required
Source code in aiorp/response.py 
26
27
28
29
30
31
32
33
34
35
36
37
def __init__(
    self,
    in_resp: client.ClientResponse,
):
    """Initialize the proxy response object.

    Args:
        in_resp: The incoming response object.
    """
    self.in_resp: client.ClientResponse = in_resp
    self._web: web.StreamResponse | None = None
    self._content: bytes | None = None

set_response(response_type=ResponseType.BASE) async

Set the response using the given response type.

Parameters:

Name Type Description Default
response_type ResponseType

The type of response to set.

 BASE

Returns:

Type Description
StreamResponse | Response

The set web response.

Raises:

Type Description
ValueError

When attempted to set the response a second time.
Source code in aiorp/response.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
async def set_response(
    self, response_type: ResponseType = ResponseType.BASE
) -> StreamResponse | Response:
    """Set the response using the given response type.

    Args:
        response_type: The type of response to set.

    Returns:
        The set web response.

    Raises:
        ValueError: When attempted to set the response a second time.
    """
    if self._web is not None:
        raise ValueError("Response can only be set once")
    if response_type == ResponseType.STREAM:
        self._web = await self._get_stream_response()
    else:
        self._web = await self._get_base_response()
    return self._web