[FIXED] How to create await-able object in python with specific methods

Issue

There is aiohttp based server that looks like:

import aiohttp
routes = web.RouteTableDef()
@routes.post('/route1')
async def route1(request):
    req_json = await request.json()
    step1 = func1(req_json)
    step2 = func2(step1)
    return web.Response(body=step2, status=200)

I want to test that server locally without running whole server, in particular step1 and step2 functions in route1.

So I do:

import asyncio
from myserver import route1
json_request = {"some": "data"}
loop = asyncio.get_event_loop()
loop.run_until_complete(route1(json_request))

Problem is, I have to replace req_json = await request.json() with req_json = request when run local tests. How to create awaitable variable with .json() method to use instead of json_request dict?

Solution

For this line to work: await request.json(), you don't actually need the request object to be awaitable. It can be a normal object with a .json() attribute which "is" an awaitable object. A coroutine would be a reasonable choice.

Like this:

import asyncio
from myserver import route1

class CustomRequest:
    def __init__(self, data: dict) -> None:
        self.data = data

    async def json(self) -> dict:
        return self.data
        
json_request = {"some": "data"}
request_object = CustomRequest(json_request)
asyncio.run(route1(request_object))

Answered By - S.B