aiohttp client methods being called from aiohttp server cause RuntimeError

Question

I'm trying to create a payment microservice with following layout:

The client side is working fine until it comes to use aiohttp client methods in aiohttp server side.

I have a class which creates an aiohttp.ClientSession and provides a different methods use aiohttp.ClientSession instance context manager to perform requests. If I init a class' instance and use its methods within a same module (asyncio.run()) everything works without warnings. But once I import my class to aiohttp server routers module, init aiohttp client class and call its methods within "views" it causes RuntimeError: Timeout context manager should be used inside a task. I guess the exception caused by some wrong task execution wrom aiohttp server existing eventloop but can't find the right implementation of client+server pairing.

My aiohttp client representation class:

class MonoAsyncClient:
    """ Client for performing HTTP requests to bank API asynchronously """
    ROOT_URL = bank_settings.bank_config.root_bank_url

    def __init__(self) -> None:
        self.headers = {
            "X-Token": bank_settings.bank_config.token
        }
        self._session = ClientSession(headers=self.headers)

    async def __aenter__(self) -> "MonoAsyncClient":
        return self

    async def __aexit__(
        self,
        exc_type: Exception,
        exc_val: TracebackException,
        traceback: TracebackType,
    ) -> None:
        await self.close()

    async def close(self) -> None:
        await self._session.close()

    async def create_payment(
            self,
            payment_data: NewPaymentSchema
    ) -> NewPaymentResponseSchema:
        """ Performs POST request to create new payment bill. """
        input_data = payment_data.model_dump_json(by_alias=True)
        async with self._session.post(
                url=self.ROOT_URL + "/invoice/create",
                data=input_data
        ) as response:
            response_data = await response.json()
            valid_data = NewPaymentResponseSchema.model_validate(response_data)
            return valid_data

Example of aiohttp server view :

routes = web.RouteTableDef()

mono_client = MonoAsyncClient()


@routes.post("/invoices/create")
async def create_payment(request: Request):
    """ View to create a new payment invoice. """
    # retrieving POST data body
    data = await request.post()
    try:
        validated_data = NewPaymentSchema.model_validate(data)
    except ValidationError as e:
        return e.json()
    else:
        bank_api_response = await mono_client.create_payment(validated_data)
        return bank_api_response.model_dump_json()

And aiohttp server app:

main_app = web.Application()
main_app.add_subapp(
    prefix=bank_settings.bank_config.root_server_v1_url,
    subapp=bank_app.v1_app
)


if __name__ == "__main__":
    web.run_app(
        app=main_app,
        host=settings.run.host,
        port=settings.run.port
    )

Answer 1

Finally I found a solution. All success credits to @Sam Bull for guiding to a right vector.

First of all I've shifted self._session init to a separate async method:

class SomeClient:

    def __init__(self) -> None:
        self.headers = {
            "X-Token": "token string"
        },
        self._session = None
    
    async def get_session(self):
        if self._session is None:
            self._session = ClientSession(headers=self.headers)
        return self._session

To init and combine aiohttp client with an existing aiohttp server app I've added an aiohttp client instance to app (e.g web.Application() instance). Example:

app = web.Application()
app["client"] = SomeClientClass()

Then the client class available from views/handlers:

@app.post("/foo")
async def view(request: Request) -> Response:
    client = request.app["client"]
    response = await client.method() # <- perfroms some request method from client class

Particulary in my case I had slightly different code due to project structure: The app's view whithin which the client being called is a nested app. In this case I had to "attach" a client instance to the "sub_app" where the client is to be called rather than "root" app.

# project/apps.py
from project.routes import routes

sub_app = web.Application()
sub_app["client"] = SomeClientClass()
# adding API routes
sub_app.add_routes(routes)

# project/main.py
from project.apps import sub_app

main_app = web.Application()
main_app.add_subapp(
    prefix="sub_app",
    subapp=sub_app
)

And then it works like a charm.