SQLAlchemy Autoflush 不适用于 Session.add答案

【问题标题】：SQLAlchemy Autoflush does not work on Session.addSQLAlchemy Autoflush 不适用于 Session.add
【发布时间】：2022-01-15 11:41:52
【问题描述】：

我想在数据库中保存新对象并在不调用session.flush() 的情况下接收新实体的自动增量 ID。我尝试了autoflush=True(docs) 选项但没有运气。

这是我的代码。

from sqlalchemy import create_engine, Column, Integer, String, Boolean
from sqlalchemy.orm import sessionmaker, as_declarative

engine = create_engine("postgresql://postgres:@localhost/postgres")
Session = sessionmaker(autocommit=False, autoflush=True, bind=engine)
session = Session()
print(session.autoflush)  # -> True

@as_declarative()
class Base:
    __name__: str

class UserDb(Base):
    __tablename__ = "user"
    id = Column(Integer, primary_key=True)
    username: str = Column(String, unique=True)
    hashed_password: str = Column(String)
    is_active: bool = Column(Boolean, default=True)
    is_superuser: bool = Column(Boolean, default=False)


user = UserDb(
    username="username",
    hashed_password="password",
)
session.add(user)
# Here I want to have user.id
print(user.id)  # -> None
session.flush()
print(user.id)  # -> 12

有没有办法在 SQLAlchemy 中实现这种行为？

环境：python 3.9.7，SQLAlchemy 1.4.27

【问题讨论】：

引用您引用的文档：“当 [autoflush is] True 时，所有查询操作都会在继续之前向此 Session 发出 Session.flush() 调用。” （强调我的。）所以session.add() 不会被刷新，直到会话上发生下一个“查询操作”。并且您需要刷新才能从服务器获取自动增量 ID 值。这就是它的工作原理。
@GordThompson 是的，我知道autoflush 无法解决我的问题。但我不明白为什么 SQLAlchemy 是以这种方式设计的。每次添加后调用flush很不方便。我是 SQLAlchemy 的新手，通常使用 DjangoORM，所以也许我不明白。
为什么需要实际的ID值？您已经在会话中拥有该对象（身份映射），因此您可以继续使用它，直到您想要提交为止。
可能会绊倒新用户的一件事是，您无需知道 ID 即可在实体之间建立关系。如果使用 relationships API，您可以将其留给 SQLA。
@IljaEverilä - 是的，这就是我的暗示。

标签： python postgresql sqlalchemy

【解决方案1】：

我敢肯定，这是一种反模式，但我已经通过猴子修补 SQLAlchemy 实现了预期的行为

from sqlalchemy.orm import Session


def add_decorator(method):  # type: ignore
    """
    Flush session to Database for Create/Update queries
    Useful for instant retrieving Autoincrement IDs

    So instead
    session.add(obj)
    session.flush()

    We can write just
    session.add(obj)

    But this solution requires more DB calls.
    To optimize code with large amount of session.add calls use

    with session.no_autoflush:
        ...
    """

    def wrapper(self, instance, _warn=True):  # type: ignore
        method(self, instance, _warn)
        if self.autoflush:
            self.flush()

    return wrapper


def patch() -> None:
    Session.add = add_decorator(Session.add)  # type: ignore

然后在需要开启autoflush的任何地方拨打patch()

【讨论】：