Питон | RuntimeError: Задача подключила Future к другому циклу | Сеанс SqlAlchemy ⇐ Python

[Anonymous]

Проект основан на FASTAPI, база данных PostgreSQL, все поднимается через Docker. Конец каждого теста просто очистит базу данных с помощью отката. Я пишу тесты для userRepository (принимает сеанс). Почему и как происходит эта ошибка - понятия не имею. Я поделюсь всеми дополнительными файлами из проекта, чтобы решить проблему, заранее спасибо! < /P>

Код: Выделить всё

@pytest.fixture(scope="session")
def config() -> Config:
return Config()

@pytest.fixture(scope="session")
def event_loop() -> Generator[AbstractEventLoop, None, None]:
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()

@pytest.fixture(scope="session")
def rdbms_config(config: Config) -> RDBConfig:
return config.RDB.model_copy(update={"dsn": str(config.RDB.dsn) + "_test"})

@pytest.fixture(scope="session")
async def database(
event_loop: AbstractEventLoop,
rdbms_config: RDBConfig,
) -> AsyncGenerator[AsyncEngine, None]:
await create_database(rdbms_config.dsn)

engine = create_async_engine(rdbms_config.dsn, poolclass=NullPool)

async with engine.begin() as conn:
await conn.run_sync(apply_migrations)

await engine.dispose()
try:
yield engine
finally:
await drop_database(rdbms_config.dsn)

@pytest.fixture(scope="session")
async def sqla_database(rdbms_config: RDBConfig, database: None) -> SqlAlchemyDatabase:
return SqlAlchemyDatabase(rdbms_config)

@pytest.fixture(scope="session")
async def sqla_engine(sqla_database: SqlAlchemyDatabase) -> AsyncEngine:
yield sqla_database.engine
await sqla_database.engine.dispose()

@pytest.fixture(scope="function")
async def session_factory(
database: AsyncEngine,
sqla_database: SqlAlchemyDatabase,
) -> async_sessionmaker[AsyncSession]:

connection = await database.connect()

yield async_sessionmaker(
bind=connection,
expire_on_commit=False,
join_transaction_mode="create_savepoint",
)

await connection.close()

@pytest.fixture(scope="function")
async def session(session_factory: async_sessionmaker[AsyncSession]) -> AsyncSession:
async with session_factory() as session:
try:
yield session
finally:
await session.rollback()
await session.close()

Вот сам тест. Никакой логики, просто поиск ошибки, ошибка связана с сессией, а вот почему и как - вопрос.
async def test_get_er_by_email(session) -> None:
await session.execute(text("SELECT 1"))
assert 1
< /code>
Точка состоит в том, что когда вы изменяете Scope = "Session" Session_Factory и Session на Scope = "function", он работает. Но работа заканчивается в точке, когда я использую Pytest для проверки ошибки из репозитория, после чего у меня есть откат, и дальнейшие тесты не проходят. < /P>
ожидание было следующим: Запустите асинхронные тесты, подключите тестовую базу данных, возьмите его сеанс и перенесите его из репозитория, который принимает этот сеанс.>

Подробнее здесь: https://stackoverflow.com/questions/793 ... oop-sqlalc