Skip to content

ProxyContext

Proxy options used to configure the proxy handler.

This class manages the context for proxy operations, including the target URL, session management, and request/response handling.

Parameters:

Name Type Description Default
url URL

The target URL to proxy requests to.

 required
session_factory SessionFactory | None

Optional factory function to create client sessions. If not provided, defaults to aiohttp.ClientSession.

 None
state dict | None

Optional state object to store additional context data.

 None
Source code in aiorp/context.py 
 14
 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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
class ProxyContext:
    """Proxy options used to configure the proxy handler.

    This class manages the context for proxy operations, including the target URL,
    session management, and request/response handling.

    Args:
        url: The target URL to proxy requests to.
        session_factory: Optional factory function to create client sessions.
            If not provided, defaults to aiohttp.ClientSession.
        state: Optional state object to store additional context data.
    """

    def __init__(
        self,
        url: URL,
        session_factory: SessionFactory | None = None,
        state: dict | None = None,
    ):
        self.url: URL = url
        self.state: dict | None = state
        self.session_factory: SessionFactory = session_factory or ClientSession
        self._request: ProxyRequest | None = None
        self._response: ProxyResponse | None = None
        self._ws_source: web.WebSocketResponse | None = None
        self._ws_target: client.ClientWebSocketResponse | None = None
        self._session: ClientSession | None = None

    def __copy__(self) -> "ProxyContext":
        """Copy the proxy context

        Shares the session object, but creates a new instance for the state.

        Returns:
            A ProxyContext instance with a new instance of state,
                but the same session object.
        """
        ctx = ProxyContext(
            url=self.url,  # Thread-safe design, always returns a new instance
            state={**self.state} if self.state else None,
            session_factory=self.session_factory,
        )
        # Set the session in case it is already there
        ctx._session = self._session
        return ctx

    @property
    def ws_source(self) -> WebSocketResponse | None:
        """WebSocketResponse in charge of handling the server side socket with the client.

        Returns:
            The WebSocketResponse
        """
        return self._ws_source

    @property
    def ws_target(self) -> ClientWebSocketResponse | None:
        """ClientWebSocketResponse in charge of handling the client side socket
        with the target server.

        Returns:
            The ClientWebSocketResponse
        """
        return self._ws_target

    @property
    def response(self) -> ProxyResponse:
        """Get the current proxy response.

        Returns:
            The current ProxyResponse object.

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

    @property
    def request(self) -> ProxyRequest:
        """Get the current proxy request.

        Returns:
            The current ProxyRequest object.

        Raises:
            ValueError: If the request has not been set yet.
        """
        if self._request is None:
            raise ValueError("Request is not yet set")
        return self._request

    def set_request(self, request: web.Request):
        """Set the current proxy request.

        Args:
            request: The incoming web request to proxy.
        """
        self._request = ProxyRequest(
            url=self.url,
            in_req=request,
        )

    def set_response(self, response: client.ClientResponse):
        """Set the current proxy response.

        Args:
            response: The response from the target server.
        """
        self._response = ProxyResponse(in_resp=response)

    @property
    def session(self) -> ClientSession:
        """Get the session object, creating it if necessary.

        Returns:
            An active ClientSession instance.

        Note:
            If the session is closed or doesn't exist, a new one will be created
            using the session factory.
        """
        self.start_session()
        return self._session

    def start_session(self):
        """Build the session using the factory"""
        if self._session is None or self._session.closed:
            self._session = self.session_factory()

    async def close_session(self):
        """Close the session object.

        This method properly closes the current session and cleans up resources.
        """
        if self._session is not None:
            await self._session.close()
        self._session = None

    def set_socket_pair(
        self, ws_source: WebSocketResponse, ws_target: ClientWebSocketResponse
    ):
        """Set the socket pair used for tunneling messages

        Args:
            ws_source: The WebSocketResponse to set
            ws_target: The ClientWebSocketResponse to set
        """
        self._ws_source = ws_source
        self._ws_target = ws_target

    async def terminate_sockets(self):
        """Terminate the sockets if any are set"""
        if self._ws_source and self._ws_target:
            await self._ws_source.close()
            await self._ws_target.close()

request property

Get the current proxy request.

Returns:

Type Description
ProxyRequest

The current ProxyRequest object.

Raises:

Type Description
ValueError

If the request has not been set yet.

response property

Get the current proxy response.

Returns:

Type Description
ProxyResponse

The current ProxyResponse object.

Raises:

Type Description
ValueError

If the response has not been set yet.

session property

Get the session object, creating it if necessary.

Returns:

Type Description
ClientSession

An active ClientSession instance.
Note

If the session is closed or doesn't exist, a new one will be created using the session factory.

ws_source property

WebSocketResponse in charge of handling the server side socket with the client.

Returns:

Type Description
WebSocketResponse | None

The WebSocketResponse

ws_target property

ClientWebSocketResponse in charge of handling the client side socket with the target server.

Returns:

Type Description
ClientWebSocketResponse | None

The ClientWebSocketResponse

__copy__()

Copy the proxy context

Shares the session object, but creates a new instance for the state.

Returns:

Type Description
ProxyContext

A ProxyContext instance with a new instance of state, but the same session object.
Source code in aiorp/context.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def __copy__(self) -> "ProxyContext":
    """Copy the proxy context

    Shares the session object, but creates a new instance for the state.

    Returns:
        A ProxyContext instance with a new instance of state,
            but the same session object.
    """
    ctx = ProxyContext(
        url=self.url,  # Thread-safe design, always returns a new instance
        state={**self.state} if self.state else None,
        session_factory=self.session_factory,
    )
    # Set the session in case it is already there
    ctx._session = self._session
    return ctx

close_session() async

Close the session object.

This method properly closes the current session and cleans up resources.

Source code in aiorp/context.py 
145
146
147
148
149
150
151
152
async def close_session(self):
    """Close the session object.

    This method properly closes the current session and cleans up resources.
    """
    if self._session is not None:
        await self._session.close()
    self._session = None

set_request(request)

Set the current proxy request.

Parameters:

Name Type Description Default
request Request

The incoming web request to proxy.

 required
Source code in aiorp/context.py 
107
108
109
110
111
112
113
114
115
116
def set_request(self, request: web.Request):
    """Set the current proxy request.

    Args:
        request: The incoming web request to proxy.
    """
    self._request = ProxyRequest(
        url=self.url,
        in_req=request,
    )

set_response(response)

Set the current proxy response.

Parameters:

Name Type Description Default
response ClientResponse

The response from the target server.

 required
Source code in aiorp/context.py 
118
119
120
121
122
123
124
def set_response(self, response: client.ClientResponse):
    """Set the current proxy response.

    Args:
        response: The response from the target server.
    """
    self._response = ProxyResponse(in_resp=response)

set_socket_pair(ws_source, ws_target)

Set the socket pair used for tunneling messages

Parameters:

Name Type Description Default
ws_source WebSocketResponse

The WebSocketResponse to set

 required
ws_target ClientWebSocketResponse

The ClientWebSocketResponse to set

 required
Source code in aiorp/context.py 
154
155
156
157
158
159
160
161
162
163
164
def set_socket_pair(
    self, ws_source: WebSocketResponse, ws_target: ClientWebSocketResponse
):
    """Set the socket pair used for tunneling messages

    Args:
        ws_source: The WebSocketResponse to set
        ws_target: The ClientWebSocketResponse to set
    """
    self._ws_source = ws_source
    self._ws_target = ws_target

start_session()

Build the session using the factory

Source code in aiorp/context.py 
140
141
142
143
def start_session(self):
    """Build the session using the factory"""
    if self._session is None or self._session.closed:
        self._session = self.session_factory()

terminate_sockets() async

Terminate the sockets if any are set

Source code in aiorp/context.py 
166
167
168
169
170
async def terminate_sockets(self):
    """Terminate the sockets if any are set"""
    if self._ws_source and self._ws_target:
        await self._ws_source.close()
        await self._ws_target.close()