[Python] How to implement a transactional decorator in FastAPI + SQLAlchemy - with reviewing other approaches

• fastapi-boilerplate

  1 880 6.7 Python

  FastAPI boilerplate for real world production (by teamhide)

from typing import Optional from contextvars import ContextVar from sqlalchemy.ext.asyncio import ( create_async_engine, async_scoped_session, async_sessionmaker, AsyncSession, ) from ..config import config # some hints from: https://github.com/teamhide/fastapi-boilerplate/blob/master/core/db/session.py db_session_context: ContextVar[Optional[int]] = ContextVar( "db_session_context", default=None ) engine = create_async_engine(url=config.DB_URL) def get_db_session_context() -> int: session_id = db_session_context.get() if not session_id: raise ValueError("Currently no session is available") return session_id def set_db_session_context(*, session_id: int) -> None: db_session_context.set(session_id) AsyncScopedSession = async_scoped_session( session_factory=async_sessionmaker(bind=engine, autoflush=False, autocommit=False), scopefunc=get_db_session_context, ) def get_current_session() -> AsyncSession: return AsyncScopedSession()

• InfluxDB

  www.influxdata.com featured

  Power Real-Time Data Analytics at Scale. Get real-time insights from all types of time series data with InfluxDB. Ingest, query, and analyze billions of data points in real-time with unbounded cardinality.

  

Suggest a related project