The problem is, when I use multiple repositories in an endpoint, a new database session is created for every repository. So when there is an error in one of the repositories, there is no way to rollback all changes. Is there a way to create a single session in a request context and share it across repositories?
\nI have thought about overriding and resetting the provider in middlewares. But I think it won't work because the container instance is shared between requests.
","upvoteCount":2,"answerCount":7,"acceptedAnswer":{"@type":"Answer","text":"Sorry for the super late answer @SarloAkrobata, I had some really busy workdays lately. Here is the solution I found:
\nFirst, I tried to use ContextLocalSingleton as suggested by @rmk135 , but I couldn't manage to use the same session for the endpoint, dependencies and middlewares. I don't remember exactly, but I think FastAPI was executing middlewares in a separate context. My alternate solution was to create the context manually.
\n# database.py looks roughly like this\n\n_request_id_ctx_var: contextvars.ContextVar[str] = contextvars.ContextVar('request_id_ctx')\n\ndef get_request_id() -> str:\n return _request_id_ctx_var.get()\n \n@contextmanager\ndef db_context(identifier: str):\n ctx_token = _request_id_ctx_var.set(identifier)\n yield\n _request_id_ctx_var.reset(ctx_token)\n\nclass Database:\n def __init__(self, db_url: str, pool_size: int) -> None:\n self.db_url = db_url\n self._engine = create_async_engine(\n self.db_url,\n echo=False,\n pool_size=pool_size,\n )\n self._create_factory()\n\n def _create_factory(self, bind: Union[Connection, Engine, None] = None) -> None:\n bind = bind or self._engine\n\n self.async_session_factory = sessionmaker(\n bind=bind,\n class_=_AsyncSession,\n expire_on_commit=False,\n autoflush=False,\n )\n self.AsyncSession = async_scoped_session(self.async_session_factory, scopefunc=get_request_id)\n\n def create_session(self):\n return self.AsyncSession()\n\n# In application.py I set up middlewares to set the context\n\n@app.middleware('http')\nasync def remove_db_session(request: Request, call_next):\n response = await call_next(request)\n\n db = container.db()\n await db.AsyncSession.remove()\n\n return response\n\n@app.middleware('http')\nasync def set_db_context(request: Request, call_next):\n request_id = str(uuid.uuid4())\n\n with db_context(identifier=request_id):\n response = await call_next(request)\n\n return response\n\n# And in containers.py\nclass Container(containers.DeclarativeContainer):\n db = providers.Singleton(Database, db_url=config.database.url, pool_size=config.database.pool_size)\n db_session = providers.Callable(db.provided.create_session.call())\n\n user_repository = providers.Factory(UserRepository, session=db_session)Disclaimers:
\n- \n
- I haven't tested the performance implications of generating a uuid for every request. But I don't think it should be a big overhead \n
- The order of middlewares is important, but as far I tested, FastAPI guarentees the execution order \n
-
|
Hi, I'm using dependency-injector and SQLAlchemy with FastAPI. I have a provider that creates a new session. This provider is used in the repositories in the following way. db = providers.Singleton(Database, db_url=config.database_url)
db_session = providers.Callable(db.provided.create_session.call())
example_repository = providers.Factory(ExampleRepository, session=db_session)The problem is, when I use multiple repositories in an endpoint, a new database session is created for every repository. So when there is an error in one of the repositories, there is no way to rollback all changes. Is there a way to create a single session in a request context and share it across repositories? I have thought about overriding and resetting the provider in middlewares. But I think it won't work because the container instance is shared between requests. |
Beta Was this translation helpful? Give feedback.
Replies: 7 comments 2 replies
-
|
Hey @hakanutku , Good question. Yeah, FastAPI request context is a known pain point. Try to use Source code: https://github.com/ets-labs/python-dependency-injector/blob/master/src/dependency_injector/providers.pyx#L2997-L3047 |
Beta Was this translation helpful? Give feedback.
-
|
Also it you got a working example, could you please share it here? I could include it into Dependency Injector docs. |
Beta Was this translation helpful? Give feedback.
-
|
@hakanutku @rmk135 Hi guys, is there any update on this? |
Beta Was this translation helpful? Give feedback.
-
|
Sorry for the super late answer @SarloAkrobata, I had some really busy workdays lately. Here is the solution I found: First, I tried to use ContextLocalSingleton as suggested by @rmk135 , but I couldn't manage to use the same session for the endpoint, dependencies and middlewares. I don't remember exactly, but I think FastAPI was executing middlewares in a separate context. My alternate solution was to create the context manually. # database.py looks roughly like this
_request_id_ctx_var: contextvars.ContextVar[str] = contextvars.ContextVar('request_id_ctx')
def get_request_id() -> str:
return _request_id_ctx_var.get()
@contextmanager
def db_context(identifier: str):
ctx_token = _request_id_ctx_var.set(identifier)
yield
_request_id_ctx_var.reset(ctx_token)
class Database:
def __init__(self, db_url: str, pool_size: int) -> None:
self.db_url = db_url
self._engine = create_async_engine(
self.db_url,
echo=False,
pool_size=pool_size,
)
self._create_factory()
def _create_factory(self, bind: Union[Connection, Engine, None] = None) -> None:
bind = bind or self._engine
self.async_session_factory = sessionmaker(
bind=bind,
class_=_AsyncSession,
expire_on_commit=False,
autoflush=False,
)
self.AsyncSession = async_scoped_session(self.async_session_factory, scopefunc=get_request_id)
def create_session(self):
return self.AsyncSession()
# In application.py I set up middlewares to set the context
@app.middleware('http')
async def remove_db_session(request: Request, call_next):
response = await call_next(request)
db = container.db()
await db.AsyncSession.remove()
return response
@app.middleware('http')
async def set_db_context(request: Request, call_next):
request_id = str(uuid.uuid4())
with db_context(identifier=request_id):
response = await call_next(request)
return response
# And in containers.py
class Container(containers.DeclarativeContainer):
db = providers.Singleton(Database, db_url=config.database.url, pool_size=config.database.pool_size)
db_session = providers.Callable(db.provided.create_session.call())
user_repository = providers.Factory(UserRepository, session=db_session)Disclaimers:
|
Beta Was this translation helpful? Give feedback.
-
|
Also @rmk135, is the example I shared okay to include in the docs? If not, what should I change? |
Beta Was this translation helpful? Give feedback.
-
|
@hakanutku , what do you think about using I originally posted it here: #495 (comment) 2 potential downsides:
EDIT: Seems like this is not the correct solution since it prevent's me to override client's inside tests where I have to instantiate container and inject DB ( |
Beta Was this translation helpful? Give feedback.
-
|
hey, 2024, any news or new ideas? the How can that be achieved with FastAPI? |
Beta Was this translation helpful? Give feedback.
-
|
|
Beta Was this translation helpful? Give feedback.
-
|
Check my answer here #728 (comment) |
Beta Was this translation helpful? Give feedback.
Sorry for the super late answer @SarloAkrobata, I had some really busy workdays lately. Here is the solution I found:
First, I tried to use ContextLocalSingleton as suggested by @rmk135 , but I couldn't manage to use the same session for the endpoint, dependencies and middlewares. I don't remember exactly, but I think FastAPI was executing middlewares in a separate context. My alternate solution was to create the context manually.