Now I just import sessionmaker in dao classes and do what I need to:
class BaseDAO:
model = None
@classmethod
async def find_all(cls, **filter_by):
async with async_session() as session:
query = select(cls.model).filter_by(**filter_by)
result = await session.execute(query)
return result.scalars().all()
I know that there are two other ways:
- pass session in endpoint via Depends(get_db):
def get_db() -> Generator: # pragma: no cover
"""
Returns a generator that yields a database session
Yields:
Session: A database session object.
Raises:
Exception: If an error occurs while getting the database session.
"""
log.debug("getting database session")
db = get_local_session(SQLALCHEMY_DATABASE_URL, False)()
try:
yield db
finally: # pragma: no cover
log.debug("closing database session")
db.close() # pragma: no cover
@router.post("/register")
def register(user_register: UserRegister, db: Session = Depends(get_db)):
- The most unclear one. But I was told that it is the best way:
a. By pointing the dependency to an abstract base class rather than a concrete class. This may be a good idea from a code structuring point of view, but can be difficult to describe when we depend on third party classes
class MyProto(abc.ABC):
...
@router.get
def my_view_funcion(param: MyProto = Depends()):
...
app = FastApi()
app.dependency_overrides[MyProto] = some_session_factory
b. Pointing a dependency to a stub function that doesn’t actually do anything. This can also be used to mark dependencies of the same type, but which should have different meanings.
def get_session_stub():
raise NotImplementedError # это реально тело этой функции
@router.get
def my_view_funcion(param: Session = Depends(get_session_stub)):
...
app = FastApi()
app.dependency_overrides[get_session_stub] = some_session_factory
It would be great if you could explain to me the last strategy. If it’s good, of course 🙂