python - Sqlalchemy 加载实例事件未触发

Question

我正在尝试自定义从数据库加载的对象。

根据文档here该事件应该在对象加载时触发，但事实并非如此。

如何触发加载实例事件？这是我的代码

# test.py
from sqlalchemy import create_engine, MetaData, Table, Column, String, Integer, event
from sqlalchemy.orm import mapper, scoped_session, sessionmaker
from sqlalchemy_utils.functions import create_database, drop_database


engine = create_engine('sqlite:///sqlite3.db')
engine.connect()

metadata = MetaData(engine)

user_table = Table('user', metadata, 
    Column('id', Integer(), primary_key=True),
    Column('name', String(200), nullable=False),
)

class User:

    def __init__(self, *args, **kwargs):
        pass


def receive_load(target, context):
    print("listen for the 'load' event")


event.listen(User, 'load', receive_load)

mapper(User, user_table)
create_database(engine.url)
metadata.create_all()

session_factory = scoped_session(sessionmaker(engine),)
session = session_factory()

user = User()
user.id = 1
user.name = 'ashwin'

session.add(user)
session.flush()
session.commit()

new_session = session_factory()

obj = new_session.query(User).filter_by(id=1).first() # Event should fire here when I run this line
drop_database(engine.url)

Answer 1

我从 sqlalchemy 的创建者本人那里得到了 sqlalchemy github 问题的答案，将其重新发布到此处，以便其他人也发现它有用。

https://github.com/sqlalchemy/sqlalchemy/issues/5008

基本上，错误是我实际上并没有在代码中创建两个不同的 session ，因为 session_factory() 两次返回相同的 session ，因为它是一个线程范围的 session 。

如果我将代码分成两个单独的脚本，然后一个接一个地运行它们，我可以看到 load 事件触发。

# test1.py
from sqlalchemy import create_engine, MetaData, Table, Column, String, Integer, event
from sqlalchemy.orm import mapper, scoped_session, sessionmaker
from sqlalchemy_utils.functions import create_database, drop_database
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine('sqlite:///sqlite3.db')
engine.connect()

metadata = MetaData(engine)

user_table = Table('user', metadata, 
    Column('id', Integer(), primary_key=True),
    Column('name', String(200), nullable=False),
)

class User:

    def __init__(self, *args, **kwargs):
        pass


Base = declarative_base()

mapper(User, user_table)


create_database(engine.url)
metadata.create_all()

session_factory = scoped_session(sessionmaker(engine),)

session = session_factory()

user = User()
user.id = 1
user.name = 'ashwin'

session.add(user)
session.flush()
session.commit()

和

from sqlalchemy import create_engine, MetaData, Table, Column, String, Integer, event
from sqlalchemy.orm import mapper, scoped_session, sessionmaker
from sqlalchemy_utils.functions import create_database, drop_database
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine('sqlite:///sqlite3.db')
engine.connect()

metadata = MetaData(engine)

user_table = Table(
    'user', 
    metadata,
    Column('id', Integer(), primary_key=True),
    Column('name', String(200), nullable=False),
)


class User:

    def __init__(self, *args, **kwargs):
        pass


def receive_load(target, context):
    print("listen for the 'load' event")
    print(target)
    print(context)

def receive_refresh(target, context, only_load_props=None):
    print("listen for the 'refresh' event")


event.listen(User, 'load', receive_load)
event.listen(User, "refresh", receive_load)

Base = declarative_base()

mapper(User, user_table)


session_factory = scoped_session(sessionmaker(engine),)

new_session = session_factory()

# Event should fire here when I run this line
obj = new_session.query(User).filter_by(id=1).first()
drop_database(engine.url)

而且，如果我需要使其在同一脚本中工作，我需要调用 session_factory.remove() 来处置第一个 session 。

from sqlalchemy import create_engine, MetaData, Table, Column, String, Integer, event
from sqlalchemy.orm import mapper, scoped_session, sessionmaker
from sqlalchemy_utils.functions import create_database, drop_database
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine('sqlite:///sqlite3.db')
engine.connect()

metadata = MetaData(engine)

user_table = Table('user', metadata,
    Column('id', Integer(), primary_key=True),
    Column('name', String(200), nullable=False),
)


class User:

    def __init__(self, *args, **kwargs):
        pass


def receive_load(target, context):
    print("listen for the 'load' event")
    print(target)
    print(context)


def receive_refresh(target, context, only_load_props=None):
    print("listen for the 'refresh' event")


event.listen(User, 'load', receive_load)
event.listen(User, "refresh", receive_load)

Base = declarative_base()

mapper(User, user_table)


create_database(engine.url)
metadata.create_all()

session_factory = scoped_session(sessionmaker(engine),)

session_factory.remove()

session = session_factory()

user = User()
user.id = 1
user.name = 'ashwin'

session.add(user)
session.flush()
session.commit()

session_factory = scoped_session(sessionmaker(engine),)

new_session = session_factory()

# Event should fire here when I run this line
obj = new_session.query(User).filter_by(id=1).first()
drop_database(engine.url)