Does graphene_sqlalchemy keep sessions open?

Bắt đầu bằng cách tái hiện hiện tượng pytest bị treo với reset_testdata(), close_all_sessions() và metadata.drop_all()/create_all() trên container Postgres dùng một lần. So sánh các request sử dụng graphene_sqlalchemy có và không có FastAPI middleware được nêu, sau đó lần theo vòng đời của session. Công việc được xem là hoàn tất khi đã xác định nguyên nhân của session còn tồn tại và hành vi cleanup dự kiến được ghi lại hoặc được bao phủ bằng một regression test.

tl;dr

Does graphene_sqlalchemy keep sessions open? I'm having the problem that Postgres doesnt run drop_all() in my test suite.

Explanation

Hi, I have a fastAPI app and I'm using graphene_sqlalchemy to generate an API from database models defined on the declarative base of SQLAlchemy. The database is postgres. For testing I am running some pytests against the API while it's running. So, I start a throw-away postgres container, then I start the API app, then I run pytest.

The tests include a reset_testdata() function which also uses the SQLAlchemy ORM. It looks like:

def reset_testdata():
    close_all_sessions()  # from sqlalchemy.orm.session
    models.Base.metadata.drop_all(bind=engine)
    models.Base.metadata.create_all(bind=engine)

    db.add_all([
        # ... add stuff
    ])

    db.commit()
    db.close()

I noticed, that once reset_testdata() is used, the pytest process hangs. No errors, not able to Ctrl+C, it just waits. I had this issue before and it usually stems from Postgres not allowing stuff like drop_all() if there are still active sessions. After some trying out I found that queries from my app (which are implemented by graphene_sqlalchemy) seem to keep a session open.
(I used this example in my app: https://docs.graphene-python.org/projects/sqlalchemy/en/latest/tutorial/#defining-our-models)

I tested this, and basically I can circumvent this problem by adding a middleware that closes all sessions after every request.

@app.middleware('http')
async def close_sessions(request: Request, call_next):
    try:
        response = await call_next(request)
    finally:
        close_all_sessions()
    return response

I wonder whether I am missing something here? Any experience with this issue (if it is even an issue)?