modern-python / fastapi-sqlalchemy-template Goto Github PK

task -l  # list of tasks with descriptions

Contributors

Stargazers

Forkers

h0rn3t ansrivas minhtran-agilityio apereiracvo edd34 jroquette wpkock hu-milize zxvum hashtegui bnugent2 ricardoltm

# ./server.py
from contextvars import ContextVar 

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import sessionmaker

_base_model_session_ctx = ContextVar("session")

@app.middleware("request")
async def inject_session(request):
    request.ctx.session = sessionmaker(bind, AsyncSession, expire_on_commit=False)()
    request.ctx.session_ctx_token = _base_model_session_ctx.set(request.ctx.session)


@app.middleware("response")
async def close_session(request, response):
    if hasattr(request.ctx, "session_ctx_token"):
        _base_model_session_ctx.reset(request.ctx.session_ctx_token)
        await request.ctx.session.close()