SQLAlchemy是否有相当于Django的get_or_create？

Question

我想从数据库中获取一个对象(如果它已经存在)(基于提供的参数),或者如果不存在则创建它.

Django get_or_create(或来源)这样做.SQLAlchemy中是否有等效的快捷方式？

我目前正在明确地写出这样的:

def get_or_create_instrument(session, serial_number):
    instrument = session.query(Instrument).filter_by(serial_number=serial_number).first()
    if instrument:
        return instrument
    else:
        instrument = Instrument(serial_number)
        session.add(instrument)
        return instrument

Answer 1

根据@WoLpH的解决方案,这是适合我的代码(简单版本):

def get_or_create(session, model, **kwargs):
    instance = session.query(model).filter_by(**kwargs).first()
    if instance:
        return instance
    else:
        instance = model(**kwargs)
        session.add(instance)
        session.commit()
        return instance

有了这个,我就可以get_or_create我的模型的任何对象.

假设我的模型对象是:

class Country(Base):
    __tablename__ = 'countries'
    id = Column(Integer, primary_key=True)
    name = Column(String, unique=True)

为了获取或创建我的对象,我写道:

myCountry = get_or_create(session, Country, name=countryName)

Answer 2

这基本上是这样做的方式,AFAIK没有随时可用的快捷方式.

你可以概括它的当然:

def get_or_create(session, model, defaults=None, **kwargs):
    instance = session.query(model).filter_by(**kwargs).first()
    if instance:
        return instance, False
    else:
        params = dict((k, v) for k, v in kwargs.iteritems() if not isinstance(v, ClauseElement))
        params.update(defaults or {})
        instance = model(**params)
        session.add(instance)
        return instance, True

Answer 3

我一直在玩这个问题,最终得到了一个相当强大的解决方案:

def get_one_or_create(session,
                      model,
                      create_method='',
                      create_method_kwargs=None,
                      **kwargs):
    try:
        return session.query(model).filter_by(**kwargs).one(), False
    except NoResultFound:
        kwargs.update(create_method_kwargs or {})
        created = getattr(model, create_method, model)(**kwargs)
        try:
            session.add(created)
            session.flush()
            return created, True
        except IntegrityError:
            session.rollback()
            return session.query(model).filter_by(**kwargs).one(), False

我刚刚写了一篇关于所有细节的相当广泛的博客文章,但是有一些关于我为什么使用它的想法.

1. 它将解包为一个元组,告诉您对象是否存在.这在您的工作流程中通常很有用.

2. 该函数提供了处理@classmethod装饰创建函数(以及特定于它们的属性)的功能.

3. 当您有多个进程连接到数据存储区时,该解决方案可以防止竞争条件.

编辑:我已经改变session.commit()到session.flush()在解释这个博客帖子.请注意,这些决策特定于所使用的数据存储区(在本例中为Postgres).

编辑2:我已经使用{}作为函数中的默认值进行了更新,因为这是典型的Python问题.谢谢你的评论,奈杰尔!如果您对此问题感到好奇,请查看此StackOverflow问题和此博客文章.

Answer 4

改进版的erik的优秀答案

def get_one_or_create(session,
                      model,
                      create_method='',
                      create_method_kwargs=None,
                      **kwargs):
    try:
        return session.query(model).filter_by(**kwargs).one(), True
    except NoResultFound:
        kwargs.update(create_method_kwargs or {})
        try:
            with session.begin_nested():
                created = getattr(model, create_method, model)(**kwargs)
                session.add(created)
            return created, False
        except IntegrityError:
            return session.query(model).filter_by(**kwargs).one(), True

• 使用嵌套事务仅回滚添加新项而不是回滚所有项(请参阅此答案以使用SQLite嵌套事务)
• 移动create_method.如果创建的对象具有关系并且通过这些关系为其分配成员,则会自动将其添加到会话中.例如,创建一个book,具有user_id和user作为对应的关系,然后在book.user=<user object>里面做create_method会添加book到会话.这意味着create_method必须在内部with才能从最终的回滚中受益.请注意,begin_nested自动触发刷新.

请注意,如果使用MySQL,则必须将事务隔离级别设置为READ COMMITTED而不是REPEATABLE READ为此工作.Django的get_or_create(和这里)使用相同的策略,另请参阅Django 文档.

Answer 5

这个SQLALchemy配方做得很好,很优雅.

要做的第一件事是定义一个给予Session使用的函数,并将字典与Session()相关联,Session()跟踪当前唯一键.

def _unique(session, cls, hashfunc, queryfunc, constructor, arg, kw):
    cache = getattr(session, '_unique_cache', None)
    if cache is None:
        session._unique_cache = cache = {}

    key = (cls, hashfunc(*arg, **kw))
    if key in cache:
        return cache[key]
    else:
        with session.no_autoflush:
            q = session.query(cls)
            q = queryfunc(q, *arg, **kw)
            obj = q.first()
            if not obj:
                obj = constructor(*arg, **kw)
                session.add(obj)
        cache[key] = obj
        return obj

使用此功能的一个例子是mixin:

class UniqueMixin(object):
    @classmethod
    def unique_hash(cls, *arg, **kw):
        raise NotImplementedError()

    @classmethod
    def unique_filter(cls, query, *arg, **kw):
        raise NotImplementedError()

    @classmethod
    def as_unique(cls, session, *arg, **kw):
        return _unique(
                    session,
                    cls,
                    cls.unique_hash,
                    cls.unique_filter,
                    cls,
                    arg, kw
            )

最后创建唯一的get_or_create模型:

from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

engine = create_engine('sqlite://', echo=True)

Session = sessionmaker(bind=engine)

class Widget(UniqueMixin, Base):
    __tablename__ = 'widget'

    id = Column(Integer, primary_key=True)
    name = Column(String, unique=True, nullable=False)

    @classmethod
    def unique_hash(cls, name):
        return name

    @classmethod
    def unique_filter(cls, query, name):
        return query.filter(Widget.name == name)

Base.metadata.create_all(engine)

session = Session()

w1, w2, w3 = Widget.as_unique(session, name='w1'), \
                Widget.as_unique(session, name='w2'), \
                Widget.as_unique(session, name='w3')
w1b = Widget.as_unique(session, name='w1')

assert w1 is w1b
assert w2 is not w3
assert w2 is not w1

session.commit()

这个秘诀深入探讨了这个想法并提供了不同的方法,但我已经用这个方法取得了巨大的成功.