添加 prefix_with 子句以插入特定类

我怎样才能定制prefix_withSQLAlchemy 中的每个模型类，以便每个模型类都可以有不同的插入语句？

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

PS：我对 SQLAlchemy 比较陌生

ORM 没有挂钩它的生成方式insert()构造，所以你在这里能做的最好的事情就是拦截insert()的声明Tablelevel，这可能已经足够好了，因为您想对这些表进行全面的“忽略”操作，这里有一个使用类装饰器使其通用的方法。我们正在利用执行前 http://docs.sqlalchemy.org/en/rel_0_8/core/events.html#sqlalchemy.events.ConnectionEvents.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()