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

Question

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

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

PS：我对 SQLAlchemy 比较陌生

Answer 1

ORM 没有挂钩它如何生成insert()构造，因此您可以在这里做的最好的事情是拦截该级别的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()