python - 添加 prefix_with 子句以插入特定类

Question

如何自定义 SQLAlchemy 中每个模型类的 prefix_with 以便每个模型类都可以有不同的插入语句？

我实际上想要将 OR IGNORE 子句添加到某些类中。

PS:我对 SQLAlchemy 比较陌生

Answer 1

ORM 没有关于它如何生成 insert() 的钩子(Hook)构造，所以你在这里能做的最好的就是拦截 insert()在 Table 的陈述级别，这在这里可能已经足够好了，因为您想对这些表全面执行“忽略”操作，这是一个使用类装饰器使其通用的方法。我们正在使用 before_execute此处重写某些事件 insert()构造:

from sqlalchemy import event
from sqlalchemy.engine import Engine
from sqlalchemy.sql import Insert

_ignore_tables = set()

@event.listens_for(Engine, "before_execute", retval=True)
def _ignore_insert(conn, element, multiparams, params):
    if isinstance(element, Insert) and \
        element.table.name in _ignore_tables:
        element = element.prefix_with("IGNORE")
    return element, multiparams, params

def ignore_inserts(cls):
    _ignore_tables.add(cls.__table__.name)
    return cls

if __name__ == '__main__':
    from sqlalchemy import Column, Integer, create_engine
    from sqlalchemy.orm import Session
    from sqlalchemy.ext.declarative import declarative_base

    Base = declarative_base()

    class A(Base):
        __tablename__ = 'a'

        id = Column(Integer, primary_key=True)

    @ignore_inserts
    class B(Base):
        __tablename__ = 'b'

        id = Column(Integer, primary_key=True)

    @ignore_inserts
    class C(Base):
        __tablename__ = 'c'

        id = Column(Integer, primary_key=True)

    e = create_engine("mysql://scott:tiger@localhost/test", echo=True)
    Base.metadata.drop_all(e)
    Base.metadata.create_all(e)

    s = Session(e)

    s.add_all([A(), B(), C()])
    s.commit()

像那样全面使用它会让我感到紧张，这是一个不同的版本，因此您可以使用上下文管理器为具有特定 Session 的特定表设置规则:

from sqlalchemy import event
from sqlalchemy.engine import Engine
from sqlalchemy.sql import Insert
from contextlib import contextmanager

@event.listens_for(Engine, "before_execute", retval=True)
def _ignore_insert(conn, element, multiparams, params):
    if isinstance(element, Insert) and \
        'ignore_tables' in conn.info and \
        element.table.name in conn.info['ignore_tables']:
        element = element.prefix_with("IGNORE")
    return element, multiparams, params


@contextmanager
def ignore_inserts(session, names):
    conn = session.connection()
    info = conn.info   # hold onto info so we can still
                       # get to it when the Connection is closed
    previous = info.get('ignore_tables', ())
    try:
        info['ignore_tables'] = set(names)
        yield
    finally:
        info['ignore_tables'] = previous

if __name__ == '__main__':
    from sqlalchemy import Column, Integer, create_engine
    from sqlalchemy.orm import Session
    from sqlalchemy.ext.declarative import declarative_base

    Base = declarative_base()

    class A(Base):
        __tablename__ = 'a'

        id = Column(Integer, primary_key=True)

    class B(Base):
        __tablename__ = 'b'

        id = Column(Integer, primary_key=True)

    class C(Base):
        __tablename__ = 'c'

        id = Column(Integer, primary_key=True)

    e = create_engine("mysql://scott:tiger@localhost/test", echo=True)
    Base.metadata.drop_all(e)
    Base.metadata.create_all(e)

    s = Session(e)

    with ignore_inserts(s, ['b']):
        s.add_all([A(), B(), C()])
        s.commit()

    with ignore_inserts(s, ['a', 'c']):
        s.add_all([A(), B(), C()])
        s.commit()