我想从数据库中获取一个对象，如果它已经存在（基于提供的参数），则获取它，否则创建它。
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
参考@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)
commit（或至少只使用flush）。这样可以将Session控制权交给方法调用者，避免过早的提交。另外，使用one_or_none()代替first()可能会更加安全一些。 - exhuma我知道的方法就是这样，没有捷径可走。
当然，你可以将其概括：
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
这是一个更简洁的版本，使用了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() 风格的获取操作。除非您还使用 with_for_update() 或可串行化事务模式，否则仍无法完全保证不会出现竞争条件。
session.add块周围加上try...except IntegrityError: instance = session.Query(...)。 - Wolph我一直在解决这个问题，最终得出了一个相当强大的解决方案：
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
我刚刚写了一篇很详细的博客文章，但是以下是我使用它的原因:
它将解压为一个元组，告诉您对象是否存在。这通常在工作流程中非常有用。
该函数允许使用带有@classmethod修饰的创建函数（以及特定于它们的属性）。
当您有多个连接到数据存储的进程时，该解决方案可以防止竞争条件。
编辑：如此博客文章所述，我已将session.commit()更改为session.flush()。请注意，这些决策是特定于使用的数据存储（在本例中为Postgres）。
编辑2：我已更新函数中的默认值为{}，因为这是Python中的常见错误。感谢Nigel的评论！如果您想了解此常见错误，请查看此StackOverflow问题和此博客文章。
get_or_create不是线程安全的，也不是原子性的。此外，Django的get_or_create如果实例被创建则返回True标志，否则返回False标志。 - Karget_or_create函数，它几乎做了完全相同的事情。这个解决方案也返回一个True/False标志来表示对象是被创建还是被获取，并且也不是原子性的。然而，线程安全和原子更新是数据库的问题，而不是Django、Flask或SQLAlchemy的问题，在这个解决方案和Django的解决方案中，都通过数据库上的事务来解决。 - erikIntegrityError的情况应该返回False，因为这个客户端没有创建这个对象，对吗？ - kevmitcherik的出色答案的修改版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
create_method。如果创建的对象具有关系，并且通过这些关系分配成员，则会自动将其添加到会话中。例如，创建一个具有user_id和user作为相应关系的book，然后在create_method中执行book.user=<user object>将会把book添加到会话中。这意味着create_method必须在with语句内部才能从可能的回滚中受益。请注意，begin_nested会自动触发刷新。请注意，如果使用MySQL，则事务隔离级别必须设置为READ COMMITTED而不是REPEATABLE READ，否则此方法将无法正常工作。Django的get_or_create（以及此处）使用相同的策略，另请参阅Django文档。
REPEATABLE READ时，IntegrityError重新查询仍可能失败并出现NoResultFound。我能想到的最好的解决方案是在此查询之前调用session.commit()，但这也不理想，因为用户可能不希望这样做。参考答案没有这个问题，因为session.rollback()具有启动新事务的相同效果。 - kevmitchcommit可能比使用rollback更糟糕，尽管对于特定的用例来说，它是可以接受的。 - Adversuscommit()。如果我对代码的理解是正确的，这就是Django所做的。 - kevmitchREAD COMMITTED，因此看起来他们并没有尝试处理这个问题。查看源代码证实了这一点。我不确定我是否理解了你的回复，你的意思是用户应该将他/她的查询放在嵌套事务中吗？我不清楚SAVEPOINT如何影响使用REPEATABLE READ进行读取。如果没有影响，那么情况似乎无法挽救，如果有影响，那么最后一个查询可能会被嵌套？ - AdversusREPEATABLE READ中，从查询之前恢复一个SAVEPOINT会使得这个查询好像从未发生过一样。因此，我发现有必要在嵌套事务中将查询语句放在try子句中，以便在IntegrityError异常子句中的查询能够正常工作。 - kevmitch这个SQLAlchemy的技巧非常好用和优雅。
首先要做的是定义一个函数，该函数将给定一个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
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()
从语义上来说，最接近的可能是：
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通常用于确保全局数据可用，因此我尽可能早地提交。
scoped_session 创建并跟踪会话，这应该是可行的，它应该实现线程安全的会话管理（这在2014年是否存在？）。 - cowbert我稍微简化了@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
有一个Python包，其中包含 @erik 提供的解决方案以及 update_or_create() 的版本。 https://github.com/enricobarzetti/sqlalchemy_get_or_create
根据您采用的隔离级别，上述解决方案都可能无法奏效。我找到的最佳解决方案是以下形式的原始 SQL：
INSERT INTO table(f1, f2, unique_f3)
SELECT 'v1', 'v2', 'v3'
WHERE NOT EXISTS (SELECT 1 FROM table WHERE f3 = 'v3')
我经常遇到的一个问题是，当一个字段有最大长度限制（比如说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
session.merge：https://dev59.com/E2ct5IYBdhLWcg3wKqa9#12298306 - Anton Tarasenko