SQLAlchemy 是否与 Django 的 get_or_create 等效？

Question

如果它已经存在（基于提供的参数），我想从数据库中获取一个对象，如果不存在则创建它。

Django 的get_or_create（或source）就是这样做的。 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

这基本上就是这样做的方法，AFAIK 没有现成的捷径。

你当然可以概括它：

def get_or_create(session, model, defaults=None, **kwargs):
    instance = session.query(model).filter_by(**kwargs).one_or_none()
    if instance:
        return instance, False
    else:
        params = {k: v for k, v in kwargs.items() if not isinstance(v, ClauseElement)}
        params.update(defaults or {})
        instance = model(**params)
        try:
            session.add(instance)
            session.commit()
        except Exception:  # The actual exception depends on the specific database so we catch all exceptions. This is similar to the official documentation: https://docs.sqlalchemy.org/en/latest/orm/session_transaction.html
            session.rollback()
            instance = session.query(model).filter_by(**kwargs).one()
            return instance, False
        else:
            return instance, True

---

2020 年更新（仅限 Python 3.9+）

这是一个更简洁的版本，带有 Python 3.9 的 the new dict union operator (|=)

def get_or_create(session, model, defaults=None, **kwargs):
    instance = session.query(model).filter_by(**kwargs).one_or_none()
    if instance:
        return instance, False
    else:
        kwargs |= defaults or {}
        instance = model(**kwargs)
        try:
            session.add(instance)
            session.commit()
        except Exception:  # The actual exception depends on the specific database so we catch all exceptions. This is similar to the official documentation: https://docs.sqlalchemy.org/en/latest/orm/session_transaction.html
            session.rollback()
            instance = session.query(model).filter_by(**kwargs).one()
            return instance, False
        else:
            return instance, True

注意：

类似于 Django 版本，这将捕获重复的键约束和类似的错误。如果您的 get 或 create 不能保证返回单个结果，它仍然可能导致竞争条件。

要缓解其中的一些问题，您需要在 session.commit() 之后添加另一个 one_or_none() 样式提取。除非您还使用 with_for_update() 或可序列化事务模式，否则这仍然不能 100% 保证不会出现竞争条件。

Answer 2

按照@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

有了这个，我可以获取或创建我模型的任何对象。

假设我的模型对象是：

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 3

This SQLALchemy recipe 做得很好，很优雅。

要做的第一件事是定义一个函数，该函数被赋予一个 Session 以使用，并将一个字典与 Session() 相关联，该 Session() 跟踪当前的 unique 键。

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()

该配方更深入地研究了这个想法并提供了不同的方法，但我已经成功地使用了这个。

Answer 4

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

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

我刚刚写了一个fairly expansive blog post 的所有细节，但我对我使用它的原因提出了一些想法。

1. 它解压成一个元组，告诉你对象是否存在。这在您的工作流程中通常很有用。

2. 该函数使您能够使用 @classmethod 修饰的创建者函数（以及特定于它们的属性）。

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

编辑：我已将session.commit() 更改为session.flush()，如this blog post 中所述。请注意，这些决策特定于所使用的数据存储（本例中为 Postgres）。

编辑 2：我在函数中使用 {} 作为默认值进行了更新，因为这是典型的 Python 陷阱。感谢the comment，奈杰尔！如果您对此问题感到好奇，请查看this StackOverflow question 和this blog post。

Answer 5

语义上最接近的可能是：

def get_or_create(model, **kwargs):
    """SqlAlchemy implementation of Django's get_or_create.
    """
    session = Session()
    instance = session.query(model).filter_by(**kwargs).first()
    if instance:
        return instance, False
    else:
        instance = model(**kwargs)
        session.add(instance)
        session.commit()
        return instance, True

不知道在 sqlalchemy 中依赖全局定义的 Session 是多么的洁净，但是 Django 版本不接受连接所以......

返回的元组包含实例和一个布尔值，指示实例是否已创建（即，如果我们从数据库中读取实例，则为 False）。

Django 的get_or_create 经常用于确保全局数据可用，因此我会尽早提交。

Answer 6

根据您采用的隔离级别，上述解决方案均无效。 我发现的最佳解决方案是以下形式的 RAW SQL：

INSERT INTO table(f1, f2, unique_f3) 
SELECT 'v1', 'v2', 'v3' 
WHERE NOT EXISTS (SELECT 1 FROM table WHERE f3 = 'v3')

无论隔离级别和并行度如何，这都是事务安全的。

注意：为了提高效率，最好为唯一列设置一个索引。

Answer 7

erik优秀answer的修改版

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

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

请注意，如果使用 MySQL，事务隔离级别必须设置为 READ COMMITTED 而不是 REPEATABLE READ 才能正常工作。 Django 的 get_or_create（和 here）使用相同的策略，另请参见 Django documentation。

Answer 8

我稍微简化了@Kevin。避免将整个函数包装在 if/else 语句中的解决方案。这样就只有一个return，我觉得更干净：

def get_or_create(session, model, **kwargs):
    instance = session.query(model).filter_by(**kwargs).first()

    if not instance:
        instance = model(**kwargs)
        session.add(instance)

    return instance

Answer 9

有一个 Python 包，其中包含 @erik 的解决方案以及 update_or_create() 的版本。 https://github.com/enricobarzetti/sqlalchemy_get_or_create

Answer 10

我经常遇到的一个问题是，当一个字段有一个最大长度（例如，STRING(40)）并且您想执行一个带有大长度字符串的 get or create 时，上述解决方案将失败。

在上述解决方案的基础上，这是我的方法：

from sqlalchemy import Column, String

def get_or_create(self, add=True, flush=True, commit=False, **kwargs):
    """

    Get the an entity based on the kwargs or create an entity with those kwargs.

    Params:
        add: (default True) should the instance be added to the session?
        flush: (default True) flush the instance to the session?
        commit: (default False) commit the session?
        kwargs: key, value pairs of parameters to lookup/create.

    Ex: SocialPlatform.get_or_create(**{'name':'facebook'})
        returns --> existing record or, will create a new record

    ---------

    NOTE: I like to add this as a classmethod in the base class of my tables, so that
    all data models inherit the base class --> functionality is transmitted across
    all orm defined models.

    """


    # Truncate values if necessary
    for key, value in kwargs.items():

        # Only use strings
        if not isinstance(value, str):
            continue

        # Only use if it's a column
        my_col = getattr(self.__table__.columns, key)

        if not isinstance(my_col, Column):
            continue

        # Skip non strings again here
        if not isinstance(my_col.type, String):
            continue

        # Get the max length
        max_len = my_col.type.length

        if value and max_len and len(value) > max_len:

            # Update the value
            value = value[:max_len]
            kwargs[key] = value

    # -------------------------------------------------

    # Make the query...
    instance = session.query(self).filter_by(**kwargs).first()

    if instance:
        return instance

    else:
        # Max length isn't accounted for here.
        # The assumption is that auto-truncation will happen on the child-model
        # Or directtly in the db
        instance = self(**kwargs)

    # You'll usually want to add to the session
    if add:
        session.add(instance)

    # Navigate these with caution
    if add and commit:
        try:
            session.commit()
        except IntegrityError:
            session.rollback()

    elif add and flush:
        session.flush()


    return instance