Hi! I'm trying to handle SQLAlchemy sessions as a resource provider instead of using FastAPI's built-in Depends.

Let me share some code:

The database class:

class Database: """Class to handle database connections and sessions""" def __init__(self, db_url: str, **kwargs): """Initialize the Database class with a database URL and optional keyword arguments""" self.db_url = db_url self.engine = create_engine(self.db_url, **kwargs, echo=settings.DEBUG_MODE, connect_args={"check_same_thread": False}, pool_size=0) self.local_session = sessionmaker(autocommit=False, autoflush=False, bind=self.engine) @contextmanager def session(self): """Context manager to handle sessions""" session: Session = self.local_session() try: yield session except Exception as e: session.rollback() finally: session.close()

The main container where I create these resources:

def get_db_session(session_factory): with session_factory() as db: yield db class AppContainer(containers.DeclarativeContainer): """Container to serve all the containers related to the app""" # ..... other injections # Setup container for talent services db = providers.Factory(Database, db_url=settings.DB_URL) db_sess = providers.Resource(get_db_session, session_factory=db.provided.session) user_container = providers.Container( UserContainer, # and this container has "UserService" that also uses the db session dependency db_sess=db_sess, ) # .... 

So, I inject the session into the services directly (or in the future in one repository). I don't like injecting the session into the routers directly that's why I prefer not using the FastAPI Depends.

The code provided works as expected, it creates 1 session per request and then closes it, the thing is that this doesn't work in a concurrent environment, I made some local testing with locust, and once I start having many requests executing with multi-threads (I'm using sync endpoints), the sessions don't work as expected, it seems that the same session is being used in more than 1 request at the same time, so I guess this is something that we could solve if we had something like FactoryResource instead of just a Resource that works as a Singleton. I tried to implement this myself by creating a CustomProvider but I couldn't make it, so if anyone has any advice I'd appreciate it.

P.S: I checked all the related issues regarding FastAPI and SQLAlchemy sessions in this repo and tried some approaches but none of them worked correctly