根据所采用的隔离级别，上述解决方案都不起作用。 我发现的最好的解决方案是一个RAW SQL在以下形式:

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

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

注意:为了提高效率，明智的做法是为唯一的列使用INDEX。

基本上就是这么做的，没有捷径可走。

当然，你可以把它概括为:

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+ ONLY)

下面是一个简洁的版本，使用Python 3.9的新字典联合运算符(|=)

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()样式的获取。这仍然不能100%保证不出现竞争条件，除非您还使用with_for_update()或可序列化事务模式。

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

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

遵循@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)

我经常遇到的一个问题是，当一个字段有最大长度(比如STRING(40))，而你想对一个大长度的字符串执行get或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

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