sqlalchemy 事件监听加载不适用于 pytest 固定装置

Question

首先，我英语不好。请理解

问题已解决，但未找到原因。

问题是，如果您将存储在 db 中的 sprit 实例从 fixure 返回到 create_sprit，则加载事件侦听器在使用该 sprit 实例的测试函数的逻辑中不起作用

我想知道为什么会出现这个问题

event.listen()：加载“”->“”/更新插入“”->“”

---

包装	版本
快速API	0.105.0
py测试	8.0.0
sqlalchemy	2.0.23

模特与活动

class Spirit(Base):
    __tablename__ = 'spirit'

    id = Column(Integer, primary_key=True)
    type = Column(Enum(SpiritType), nullable=False)
    name = Column(String(length=50), nullable=True)
    name_ko = Column(String(length=50), nullable=True)
    unit = Column(Enum(Unit), nullable=False)
    amount = Column(Integer, nullable=False)
    cocktail_id = Column(Integer, ForeignKey('cocktail.id'))
    # cocktail = relationship("Cocktail", backref="spirits")


class Material(Base):
    __tablename__ = 'material'

    id = Column(Integer, primary_key=True)
    type = Column(Enum(MaterialType), nullable=False)
    name = Column(String(length=50), nullable=False)
    name_ko = Column(String(length=50), nullable=False)
    unit = Column(Enum(Unit), nullable=False)
    amount = Column(Integer, nullable=False)
    cocktail_id = Column(Integer, ForeignKey('cocktail.id'), nullable=True)
    # cocktail = relationship("Cocktail", backref="materials")


class Cocktail(Base):
    __tablename__ = 'cocktail'

    id = Column(Integer, primary_key=True)
    name = Column(String(length=50), nullable=False)
    name_ko = Column(String(length=50), nullable=False)
    abv = Column(Integer, nullable=False)
    skill = Column(Enum(Skill), nullable=False)
    usage_count = Column(Integer, default=0)

    spirits = relationship("Spirit", cascade="all, delete", backref="cocktail")
    materials = relationship("Material", cascade="all, delete", backref="cocktail")


# define: Event Listener function
def serdes_columns(target, action):
    print(f"serdes_columns: {action}")
    for column in class_mapper(target.__class__).columns:
        if column.key in ['name', 'name_ko']:
            value = getattr(target, column.key)
            if value is not None:
                modified_value = value.replace(" ", "_") if action == 'serialize' else value.replace("_", " ")
                setattr(target, column.key, modified_value)
                print(f"{action} {column.key}: {value} -> {modified_value}")


def after_select_listener(target, connection, **kwargs):
    serdes_columns(target, 'deserialize')


def before_insert_listener(mapper, connection, target):
    serdes_columns(target, 'serialize')


def before_update_listener(mapper, connection, target):
    serdes_columns(target, 'serialize')


for model in [Spirit, Material, Cocktail]:
    event.listen(model, 'load', after_select_listener)
    event.listen(model, 'before_insert', before_insert_listener)
    event.listen(model, 'before_update', before_update_listener)

会议测试

engine = create_engine(SQLALCHEMY_DATABASE_URL)
TestSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine, class_=TestSession)


@pytest.fixture(scope="session")
def init_db():
    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)
    yield
    # Base.metadata.drop_all(bind=engine)


@pytest.fixture(scope="function")
def db(init_db):
    session = TestSessionLocal()
    try:
        yield session
    finally:
        session.rollback()
        session.close()


@pytest.fixture(scope="function")
def client(db):
    app.dependency_overrides[get_db] = lambda: db
    with TestClient(app) as _client:
        yield _client


@pytest.fixture
def spirit(db):
    spirit_data = {
        "type": SpiritType.Whisky,
        "name": "Test Spirit",
        "name_ko": "테스트 기주",
        "unit": Unit.ml,
        "amount": 50,
        "cocktail_id": None
    }

    # ** sqlalchemy event listen load not working **
    _spirit = spirit_crud.create_spirit(db, spirit_schema.SpiritCreate(**spirit_data))
    return _spirit

    # ** but, this is work **
    # return _spirit.id
    # return spirit_schema.SpiritCreate(**spirit_data)

测试

def test_spirit_detail(client, spirit):
    response = client.get("/api/spirits/1")
    print(response.json())
    assert response.status_code == 200

结果

日志，事件监听加载不起作用

------------------------------- live log setup --------------------------------
DEBUG    2024-02-27 17:44:20 asyncio::proactor_events.py:__init__:633: Using proactor: IocpProactor
serdes_columns: serialize
serialize name: Test Spirit -> Test_Spirit
serialize name_ko: 테스트 기주 -> 테스트_기주
-------------------------------- live log call --------------------------------
INFO     2024-02-27 17:44:20 httpx::_client.py:_send_single_request:1027: HTTP Request: GET http://testserver/api/spirits/1 "HTTP/1.1 200 OK"
PASSED                                                                   [100%]
{'type': 'Whisky', 'name': 'Test_Spirit', 'name_ko': '테스트_기주', 'unit': 'ml', 'amount': 50, 'id': 1, 'cocktail_id': None}

日志，事件监听加载正常

------------------------------- live log setup --------------------------------
DEBUG    2024-02-27 17:47:30 asyncio::proactor_events.py:__init__:633: Using proactor: IocpProactor
serdes_columns: serialize
serialize name: Test Spirit -> Test_Spirit
serialize name_ko: 테스트 기주 -> 테스트_기주
-------------------------------- live log call --------------------------------
INFO     2024-02-27 17:47:30 httpx::_client.py:_send_single_request:1027: HTTP Request: GET http://testserver/api/spirits/1 "HTTP/1.1 200 OK"
PASSED                                                                   [100%]
serdes_columns: deserialize
deserialize name: Test_Spirit -> Test Spirit
deserialize name_ko: 테스트_기주 -> 테스트 기주
{'type': 'Whisky', 'name': 'Test Spirit', 'name_ko': '테스트 기주', 'unit': 'ml', 'amount': 50, 'id': 1, 'cocktail_id': None}

---

如果返回值是由fixture创建的实例以外的任何值或对象，则不会出现问题。
如果您不使用该夹具，问题就不会发生。

sqlalchemypytestfastapipython

Answer 1

原因

1. 这两个装置共享同一个数据库会话。
2. 返回在固定装置中创建的实例时，它不会从内存中删除，并且在测试功能期间继续存在于恒等映射中。
3. 因此事件监听加载不起作用。

解决方案

1. 在两个装置中使用不同的数据库会话
2. 返回除创建的实例之外的对象值