python-dependency-injector
python-dependency-injector copied to clipboard
ets-labs
Is that possible to pass the same factory dependency to all dependants?
For example, I have 1 use case, it has 3 dependencies - Session, ProductRepository, and UserRepository; repositories depend on session. Could I pass single SQLAlchemy session to all them? When I create second use case, session should be different.
I want to achieve similar behaviour to FastAPI, where single dependency resolved only once per request. I saw solution @rmk135 suggested, but I cannot deregister dependency after request in async context, because multiple requests could enter simultaneously.
Also provided example with FastAPI + SqlAlchemy is not relevant. I don't want to commit session in the end of repository's method call. I want to commit it after all repositories finished their work (somewhere in service). Committing session in repository just doesn't make sense.
Hi @AlexandrFarkas. Dependency Injector provides ContextLocalSingleton that helps to solve this problem in async frameworks. FastAPI unfortunately doesn't support async context switching, so the provider doesn't work out of the box. aiohttp, for instance, supports context switching out of the box and ContextLocalSingleton does the thing.
I hear a lot of people struggling with this issue. You're not the only who faces the exact problem. I don't have a good example at the moment, but I believe the answer to this question lays in contextvars. I'll try to book some time and get back with a working sample.
hi @rmk135 i posted in stackoverflow regarding this can u help me how to do this in console app ?
https://stackoverflow.com/questions/68931178/python-depedency-injector-passing-single-sqlalchemy-db-connection-every-request
Hi @AlexandrFarkas , no, not yet. Apologies for the delayed responses. Will try to find some time this week.
Hi, just dropping by to say that I'm struggling with a similar setup where I want to keep one session to one request. I've tried to read into contextvars, but I don't feel knowledgeable enough in the async space to contribute a solid solution here.
Update: Ok I played around with it and settled on the following solution. I'm using a singleton Database object that is responsible for managing the engine and enables us to create a session that is scoped to one request by creating a unique ID per request in the middleware.
# middleware setup
app = FastAPI()
@app.middleware("http")
@inject
async def db_session_middleware(
request: Request, call_next, db: Database = Provide[Container.db]
) -> Response:
# This call ensures that we are scoping the sessions (in my case this also commits or does a rollback on error)
with db.scoped_session():
return await call_next(request)
database.py:
from contextlib import contextmanager
from contextvars import ContextVar
from typing import Iterator, Optional
from uuid import uuid4
from loguru import logger
from sqlalchemy import create_engine
from sqlalchemy.engine import Engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session, scoped_session, sessionmaker
Base = declarative_base()
class ScopedSessionAlreadyInitialized(Exception):
pass
class Database:
"""
Access the database via this class. This enables you to use the same session
in the application based on a running scope.
The scope is set via ContextVar, so it will be unique per running async
task, which makes it compatible with FastAPI and creates a new session per
incoming web request.
"""
engine: Engine
def __init__(self, db_url: str):
self.engine = create_engine(db_url)
self._session_scope_id: ContextVar[Optional[str]] = ContextVar(
"_session_scope_id", default=None
)
self._session_factory = scoped_session(
sessionmaker(
autocommit=False,
autoflush=False,
bind=self.engine,
),
# This is the magic that tells sqlalchemy if it should reuse a session.
scopefunc=lambda: self.get_scope_id(),
)
def _format_session(self, session: Session) -> str:
return f"Session({id(session)}, scope={self.get_scope_id()}, db={id(self)})"
def get_scope_id(self) -> Optional[str]:
return self._session_scope_id.get()
@contextmanager
def scoped(self):
existing_scope_id = self.get_scope_id()
if existing_scope_id is not None:
raise ScopedSessionAlreadyInitialized(
f"There is already a scoped session running ({existing_scope_id}). Did you forget to reset?"
)
scope_id = str(uuid4())
token = self._session_scope_id.set(scope_id)
try:
yield
finally:
self._session_factory.remove()
self._session_scope_id.reset(token)
def session_factory(self) -> Session:
session = self._session_factory()
logger.trace(f"session_factory produced: {self._format_session(session)}")
return session
@contextmanager
def session(self, autocommit: bool = True) -> Iterator[Session]:
session: Session = self.session_factory()
try:
yield session
except Exception as exc:
logger.trace(f"Rollback {self._format_session(session)}")
session.rollback()
raise
else:
if autocommit:
logger.trace(f"committing {self._format_session(session)}")
session.commit()
finally:
session.close()
@contextmanager
def scoped_session(self, autocommit: bool = True) -> Iterator[Session]:
with self.scoped():
with self.session(autocommit=autocommit) as session:
yield session
And then the database is configured as singleton, but we can still inject just a session when necessary:
from dependency_injector import containers, providers
from .database import Database
class Container(containers.DeclarativeContainer):
config = providers.Configuration()
# We create a single database instance, but the object itself is capable of
# creating scoped sessions per request.
db = providers.Singleton(Database, config.db_url)
# So when another object requests a database session, then we can simply
# create one from the database object, as we can be sure it is scoped.
db_session = providers.Factory(
lambda db: db.session_factory(),
db=db,
)
# For example, this injects just the session. If running in a request, this is scoped to the request, otherwise it uses a "global" session.
# In my understanding, its important here to *not* use a providers.Singleton, but providers.Factory. Otherwise the session would leak from the first request into other requests for this repository/service.
my_repository = providers.Factory(MyRepository, db=db_session)
I'm really curious of what others think. Is this a viable approach when using FastAPI + uvicorn? I'm not sure if I'm using ContextVar and sqlalchemy's scoped_session correctly here in combination.
Am I missing cases where a session would need to be closed in order to be freed up?
I've tried to use Factory but since it's not a generator it just opens the session but never closes it (and there is no option since Factory uses return).
Could solution be dynamic container?
I've been having the same issue and figured out a way to solve it using the Unit Of Work pattern.
What I didn't like about the linked implementation is that repositories are defined and instantiated in the UOW.
I wanted to use the container for defining the repositories, like we're used to. So I came up with a hybrid approach where the container defines the repo factories, and passes them to the UOW object. The UOW then only invokes the factories with a session inside its context manager.
It looks like this:
class Session:
def commit(self):
...
def rollback(self):
...
def close(self):
...
class UserRepository:
def __init__(self, session):
self.session = session
class TaskRepository:
def __init__(self, session):
self.session = session
class Service:
def __init__(self, uow):
self.uow = uow
def do_thing(self):
with self.uow as uow:
assert (
uow.session
== uow.user_repository.session
== uow.task_repository.session
)
class Container(containers.DeclarativeContainer):
session = providers.Factory(Session)
uow = providers.Factory(
UnitOfWork,
session_factory=session.provider,
user_repository=providers.Factory(UserRepository).provider,
task_repository=providers.Factory(TaskRepository).provider,
)
service = providers.Factory(Service, uow=uow)
if __name__ == "__main__":
container = Container()
container.service().do_thing()
The modified UOW:
class AbstractUnitOfWork(abc.ABC):
def __enter__(self) -> "AbstractUnitOfWork":
return self
def __exit__(self, *args):
self.rollback()
@abc.abstractmethod
def commit(self):
raise NotImplementedError
@abc.abstractmethod
def rollback(self):
raise NotImplementedError
class UnitOfWork(AbstractUnitOfWork):
def __init__(
self,
session_factory: Callable,
**repo_factories: Dict[str, dependency_injector.providers.Factory],
):
self.session_factory = session_factory
self.repo_factories = repo_factories
def __enter__(self):
self.session = self.session_factory()
for repo_name, repo_factory in self.repo_factories.items():
self.__setattr__(
repo_name,
repo_factory(session=self.session),
)
return super().__enter__()
def __exit__(self, *args):
super().__exit__(*args)
self.session.close()
def commit(self):
self.session.commit()
def rollback(self):
self.session.rollback()
- The same session is used across repos.
- The use of a context manager ensures that the session is closed when we're done.
- No need for
ContextVars,ContextLocalSingleton, per-request objects, etc. - Repos are defined in the container.
- We get support for transactions as a bonus from using UOW.
Hope this helps someone in the future.
@thatguysimon thanks for sharing this, it looked promising so I gave it a try.
I had to interact with service/repository layer under with service.uow: statement and this seems to allow me to use one session in that service.
On the downside this only let me use a single service inside controller because using 2 services will open 2 sessions (assert service.uow.session == service2.uow.session fails - debugger also sees these two objects on different memory addresses).
Would solution be to always inject only UOW into controller (instead of separate services) and also instantiate services inside it (like it's done with repositories)? This way each service can also share the same DB session. One other downside is that one can't inject repository alone inside controller and always has to use service layer.
One other thing I've noticed is that self.rollback() is called after every request which may not be what we want?
I think in ideal case these features should also be supported:
- 1 DB session per multiple services
- UOW automatically instantiated during injection
- only instantiate services and repositories that are injected into (used by) controller
- option to inject database session and/or repository directly while keeping the same session (would be useful for prototyping)
Will write back if I find any improvements.
Hey @tad3j, Yeah sounds like the UOW needs to move one layer up in your case.
The session and repositories are initialized once we enter the context manager. So if you initialize your services from inside the with, and pass through the repos from the uow object, then they will share the same session.
Regarding rollback, if you're referring to the rollback done in __exit__ then that part is lifted from the UOW implementation I mentioned above. See no. 4 for an explanation on the rollback part.
@thatguysimon, I was thinking about it and I somehow didn't like it because I believe Container class is actually UOW and by doing that you would be adding a new one.
The way I got it working now is by using Dependency class for database_session which lets me provide single session (per incoming request) for the whole container (and repositories/services) with the use of starlette/fastapi middleware for instantiating the container:
class Container(DeclarativeContainer):
database_session = Dependency(instance_of=Session)
class ServiceContainerMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
database_session = SessionLocal()
Container(database_session=database_session)
try:
return await call_next(request)
finally:
database_session.close()
The only down side of this approach is that one has to instantiate the whole container (for each request) so he can inject database_session which is probably a bit slower than having it stored in memory for all requests. Does anyone see anything wrong with this approach?
Would using override functionality be more appropriate to use here instead of Dependency? (I tried but it was opening 2 sessions: one opened by Factory and the other one by override)
EDIT: Notice they are doing similar here (just not per request): https://github.com/ets-labs/python-dependency-injector/issues/344#issuecomment-751499569 EDIT2: There is an ongoing discussion here as well: https://github.com/ets-labs/python-dependency-injector/discussions/493
