在pytest数据库初始化fixture中调用Alembic迁移无执行效果

为FastAPI项目配置pytest fixture，目标是在内存SQLite数据库上执行迁移完成测试环境初始化。但调用Alembic的upgrade命令后无任何效果，检查发现预期的players表并未创建。Alembic目录可被正常找到并调用，执行过程无任何报错提示。手动执行迁移可正常生成SQLite文件并创建所需的players表，通过FastAPI Swagger手动测试也可正常运行，仅在pytest fixture中出现此问题。

Pytest测试文件 #

from alembic import command
from alembic.config import Config
from fastapi.testclient import TestClient
import logging
from os import environ, path
import pytest
from sqlalchemy import create_engine, StaticPool, inspect
from sqlalchemy.orm import sessionmaker

from src.main import app
from database.models import get_db
from alembic import op
import sqlalchemy as sa


LOG_PATH = path.join(path.dirname(path.abspath(__file__)), '../tests/test.log')
logging.basicConfig(
    format='%(asctime)s - %(levelname)s - %(message)s',
    filename=LOG_PATH,
    filemode='a',
    level=logging.DEBUG,
)
logging.debug('Test Initalize')

client = TestClient(app)

TEST_DATABASE_URL = 'sqlite:///:memory:'

engine = create_engine(
    TEST_DATABASE_URL,
    connect_args={
        'check_same_thread': False,
    },
    poolclass=StaticPool,
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# This method and the dependency_overide, overrides the get_db method in the main.py file
def override_get_db():
    """Override the database session for testing."""
    database = TestingSessionLocal()
    try:
        yield database
    finally:
        database.close()

app.dependency_overrides[get_db] = override_get_db


# Setup the database for testing
@pytest.fixture(scope='session', autouse=True)
def apply_migrations():
    """Setup the database for testing."""
    logging.debug('Applying migrations')
    engine = create_engine(TEST_DATABASE_URL)
    with engine.begin() as connection:
        alembic_cfg = Config('alembic.ini')
        alembic_cfg.attributes['configure_logger'] = True
        alembic_cfg.attributes['connection'] = connection
        alembic_cfg.set_main_option('script_location', './src/alembic')
        try:
            command.ensure_version(alembic_cfg)
            command.upgrade(alembic_cfg, 'head') # <- alembic upgrade here
            assert engine.dialect.has_table(engine, 'players') # <- error is found here, no table
        except Exception as e:
            logging.error(f'Exception during migration: {e}')
            raise e
        yield
        engine.dispose()

def test_create_player(apply_migrations):
    # arrange
    data = {
        'playerName': 'test_player',
        'password': 'test_password',
        'email': 'test@noplace.com',
        'isAdmin': False,
    }

    # act
    # inspector = inspect(engine)
    # tables = inspector.get_table_names()

    response = client.post(
        '/players/',
        json=data,
    )

    # assert
    assert response.status_code == 201
    response_data = response.json()
    assert response_data['playerId'] == 1
    assert response_data['playerName'] == 'test_player'
    assert response_data['password'] == 'test_password'
    assert response_data['email'] == 'test@noplace.com'
    assert response_data['isAdmin'] is False

迁移脚本 #

"""initial migration

Revision ID: 264e0a95c5f8
Revises: 
Create Date: 2023-12-30 18:53:59.435418

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision: str = '264e0a95c5f8'
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('characters',
    sa.Column('characterId', sa.Integer(), nullable=False),
    sa.Column('characterName', sa.String(), nullable=True),
    sa.Column('playerId', sa.Integer(), nullable=True),
    sa.Column('isActive', sa.Boolean(), nullable=True),
    sa.PrimaryKeyConstraint('characterId')
    )
    op.create_table('players',
    sa.Column('playerId', sa.Integer(), nullable=False),
    sa.Column('playerName', sa.String(), nullable=True),
    sa.Column('password', sa.String(), nullable=True),
    sa.Column('email', sa.String(), nullable=True),
    sa.Column('isAdmin', sa.Boolean(), nullable=True),
    sa.Column('isActive', sa.Boolean(), nullable=True),
    sa.PrimaryKeyConstraint('playerId')
    )
    op.create_table('transactions',
    sa.Column('transactionId', sa.Integer(), nullable=False),
    sa.Column('characterId', sa.Integer(), nullable=True),
    sa.Column('amount', sa.Float(precision=2), nullable=True),
    sa.Column('description', sa.String(), nullable=True),
    sa.Column('transactionDate', sa.DateTime(), nullable=True),
    sa.PrimaryKeyConstraint('transactionId')
    )
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_table('transactions')
    op.drop_table('players')
    op.drop_table('characters')
    # ### end Alembic commands ###

内容的提问来源于stack exchange，提问作者cantis