-
Notifications
You must be signed in to change notification settings - Fork 0
Open
Description
Currently, in one of my codebases, the following code gets repeated in every function.
from admin_table.application import current_user
async def my_function(param1: str, ...) -> None:
async with async_manager() as mgr:
mgr.set_security_context(user_id=current_user.user_id)
mgr.call_private_function()It would be nice to setup request-lifetime hooks in order to remove the redundant code.
We could, for example, pass request_lifespan (similarily as fastapi does) to the config, which would then wrap every request.
So the afformentioned case would be simplified to the following code
from contextvars import ContextVar
from contextlib import asynccontextmanager
from admin_table.application import current_user
mgr = ContextVar("mgr")
@asynccontextmanager
async def request_lifespan():
async with async_manager() as scope_mgr:
scope_mgr.set_security_context(user_id=current_user.user_id)
token = mgr.set(scope_mgr)
try:
yield
finally:
mgr.reset(token)
async def my_function(param1: str, ...) -> None:
mgr.get().call_private_function()Metadata
Metadata
Assignees
Labels
No labels