Добавить предложение prefix_with для вставки определенного класса

Question

Как настроить класс prefix_with каждого класса модели в SQLAlchemy, чтобы каждый из них мог иметь другую инструкцию insert?

Я действительно хочу, чтобы статья 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()