First of all, I am not good at English. Please understand
Problem solved, but no cause found.
Problem is, if you return a sprit instance stored in db from fixure to create_sprit, the load event listener does not work in the logic of the test function that uses that sprit instance
I wonder why this problem occurs
event.listen(): load "" -> " " / upsert " " -> ""
| package | version |
|---|---|
| fastapi | 0.105.0 |
| pytest | 8.0.0 |
| sqlalchemy | 2.0.23 |
Models & Event
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)
conftest
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)
Test
def test_spirit_detail(client, spirit):
response = client.get("/api/spirits/1")
print(response.json())
assert response.status_code == 200
result
log, event listen load not work
------------------------------- 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}
log, event listen load is work
------------------------------- 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}
If the return value is any value or object other than an instance created by fixture, no problem occurs.
And if you don't use that fixture, the problem won't happen.
cause
Solution plan