CODEWITHSUNDEEP
pythonsqlalchemypytestfastapi
coldblade2000
coldblade2000

Reputation: 65

Possible session deadlock when testing a FastAPI route that uses SQLAlchemy

I am making a FastAPI 0.101 server and SQLAlchemy 2.0.20, but I'm having some trouble making the tests. Essentially what I think is happening is there's a deadlock happening as both the tests (PyTest), and the code run on the route try to use the same Session that is in a yield expression. I'll show you on a route whose job is just to empty the Users table. On my test, I want to clear the table, add a User, call the endpoint and then make sure there are 0 items in the table. Aside from that, I want all this test to happen in a transaction so it theoretically doesn't affect the existing data in the database, despite the endpoint calling session.commit()

Whenever the test's execution reaches the line in test_router.py marked with "# HERE", the program hangs forever, which I'm guessing is a deadlock on the session.

Any better idea on how I can do those tests on the database? Needing to keep a transaction that is later rollbacked, despite calling commit() in the code being tested

database.py

SQLALCHEMY_DATABASE_URL = "postgresql://postgres:[email protected]:5432/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL
)
Session = sessionmaker(autocommit=False, autoflush=True, bind=engine, expire_on_commit=False)

Base = declarative_base()


# Dependency
def get_session():
    """Shares a single db session to be used throughout the service"""
    session = Session()
    print("Tried to get session")
    try:
        yield session
    finally:
        session.close()

router.py

@router.post("/reset")
async def reset(
    session: Session = Depends(get_session),
):
    """
    Clears the users table
    Returns:
        msg: Todos los datos fueron eliminados
    """
    try:
        statement = delete(User)
        print(statement)
        with session:
            session.execute(statement)
            session.commit()
    except Exception as e:
        print("ERROR: /users/reset")
        print(e)
        return JSONResponse(
            status_code=500,
            content={
                "msg": "Un error desconocido ha ocurrido"
            })

    return {"msg": "Todos los datos fueron eliminados"}

conftest.py

@pytest.fixture(scope="session")
def session() -> Generator:
    Base.metadata.create_all(bind=engine)
    connection = engine.connect()

    # begin a non-ORM transaction
    transaction = connection.begin()
    session = orm.Session(bind=connection)
    yield session
    session.close()
    transaction.rollback()
    connection.close()


@pytest.fixture(scope="session")
def client() -> Generator:
    with TestClient(app) as c:
        yield c

test_router.py

def test_reset(
        client: TestClient, session: Session
):
    with session as ses:
        ses.execute(
            delete(User)
        )
        mock_user = User(
            id=uuid.UUID("c62147cf-2e63-4508-b1ff-98f805577f2c"),
            username="user",
            email="[email protected]",
            phoneNumber="+57 300 500 2000",
            dni="10100190",
            fullName="Usuario Perez Gómez",
            passwordHash="a68f7b00815178c7996ddc88208224198a584ce22faf75b19bfeb24ed6f90a59",
            salt="ES25GfW7i4Pp1BqXtASUFXJFe9PMb_7o-2v73v3svWc",
            token="eHBL1jbhBY6GfZ96DC03BlxM38SPF3npRBceefRgnkTpByFexOe7RPPDdLCh9gejD6Fe6Kdl_s5C3Gljqh3WM2xW1IGdlZQYg"
                  "V0_v55tw_NB19oMzH2t9AjKycEDdwmqPFJVR4sZuk9MFvSGoY_vQa4Y0pwCvxhBDT1VNsDnQio",
            status=UserStatusEnum.NO_VERIFICADO,
            expireAt=datetime.datetime.now(),
            createdAt=datetime.datetime.now(),
            updateAt=datetime.datetime.now()
        )
        session.add(mock_user)
        session.commit()
        assert session.scalar(select(func.count()).select_from(User)) == 1, "Table couldn't be set up successfully"

        response = client.post("/users/reset") # HERE
        assert response.status_code == 200, "The request failed for an unknown reason"
        assert "los datos fueron eliminados" in str(response.json()['msg'])
        assert session.scalar(select(func.count()).select_from(User)) == 0, "Table reset was unsuccessful"

Upvotes: 0

Views: 514

Answers (1)

Ashkan Goleh Pour
Ashkan Goleh Pour

Reputation: 522

as my experience remove with while using dependecy injection (get_session) or make a context manager in your get_session, while you using with it means you need to use context and in your current code you are context behavior with your session while it's not a context.

i prefer create a db_manager class and create enter and exit dunder methods when you want to use context manager or use builtin module for it

Upvotes: 1

Related Questions