async generators are unimplemented

[dineshbvadhia]

The mypyc error "async generators are unimplemented" occurs when the FastApi/Starlette lifespan feature is used:

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[dict[str, str], None]:
    yield {"wow": "wow"}

Does this mean mypyc has not implemented support for AsyncGenerator? If so, is there a workaround?

[JelleZijlstra]

You can perhaps write out the AsyncContextManager directly, with __aenter__ and __aexit__ methods.

[dineshbvadhia]

@JelleZijlstra Not sure what you mean to write out the AsyncContextManager directly, with __aenter__ and __aexit__ methods? The Starlette lifespan method (and by extension FastApi) uses __aenter__ and __aexit__ but maybe you mean something else.

[JelleZijlstra]

@asynccontextmanager is essentially sugar for a class with __aenter__ and __aexit__ methods. You can instead write a class with those methods.

[dineshbvadhia]

@hauntsaninja How do you let it go dynamically typed with mypyc?

[dineshbvadhia]

@JelleZijlstra I coded a simple'ish AsyncContextManager with __aenter__and __aexit__ methods:

class CustomAsyncContextManager:
    def __init__(self, startup: Callable[[], None], cleanup: Callable[[], None]):
        self.startup = startup
        self.cleanup = cleanup
        self.dao = None

    async def __aenter__(self):
        self.dao = await startup()
        return self.dao

    async def __aexit__(self, type, value, traceback):
        await cleanup()

def startup():
    dao = Dao()
    return dao

def cleanup():
    return 

async def custom_lifespan(app: FastAPI):
    async with CustomAsyncContextManager(startup=startup, cleanup=cleanup) as cacm:
        print('cacm:', cacm.dao)

app = FastAPI(lifespan=custom_lifespan)

Which generates the error:

ERROR:    Traceback (most recent call last):
  File "...\starlette\routing.py", line 677, in lifespan
    async with self.lifespan_context(app) as maybe_state:
TypeError: 'coroutine' object does not support the asynchronous context manager protocol

What am I doing wrong here?

[dineshbvadhia]

@JelleZijlstra Setting

app = FastAPI(lifespan=asyncio.run(custom_lifespan()))

eliminates the above error.

The program enters and exits CustomAsyncContextManager and so need a yield statement. FastApi/Starlette need to yield a dictionary ({'dao' : dao} in my case) which I can sort out but not where to place it. Thanks

[dineshbvadhia]

Will be resolved when mypyc supports async generators. Closing discussion