I’m working on a project that uses FastAPI and SQLAlchemy asynchronously.
I’ve written pytest tests for this project and have successfully implemented database rollback after each test run.
I’ve found two different implementation methods, but I’m unsure about the differences between them. Are both methods correct, or is one potentially problematic?
conftest.py
# pyproject.toml
#
# pytest = "^8.3.2"
# pytest-asyncio = "==0.21.2"
# #pytest-dependency = "^0.6.0"
# pytest-order = "^1.2.1"
#
# [tool.pytest.ini_options]
# addopts = "-s"
# asyncio_mode = "auto"
import asyncio
from urllib.parse import urlparse
import pytest
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from config import settings
from depends.db import async_session as api_async_session
from main import app
url = urlparse(settings.db)._replace(scheme="postgresql+asyncpg").geturl()
async_db = create_async_engine(url, echo=False, pool_size=50)
async_session_factory = async_sessionmaker(bind=async_db)
@pytest.fixture(scope="session")
def event_loop(request):
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()
@pytest.fixture
async def async_session():
async with async_db.connect() as connection:
async with connection.begin() as transaction:
async with async_session_factory(bind=connection) as s:
app.dependency_overrides[api_async_session] = lambda: s
yield s
await transaction.rollback()
@pytest.fixture
async def async_session2():
async with async_db.connect() as connection:
transaction = await connection.begin()
async with async_session_factory(bind=connection, join_transaction_mode="create_savepoint") as s:
app.dependency_overrides[api_async_session] = lambda: s
yield s
await transaction.rollback()
I’ve also checked the official documentation for create_savepoint
, but it’s too difficult to understand. Even after looking into the souced code of AsyncTransaction’s __aenter__
method, I’m still uncertain.