SqlAlchemy 2.0 中文文档（二十四）（1）

class SomeClass(Base):
    __tablename__ = "some_table"
    # ...
    value = mapped_column(Integer)
someobject = session.get(SomeClass, 5)
# set 'value' attribute to a SQL expression adding one
someobject.value = SomeClass.value + 1
# issues "UPDATE some_table SET value=value+1"
session.commit()

class Foo(Base):
    __tablename__ = "foo"
    pk = mapped_column(Integer, primary_key=True)
    bar = mapped_column(Integer)
e = create_engine("postgresql+psycopg2://scott:tiger@localhost/test", echo=True)
Base.metadata.create_all(e)
session = Session(e)
foo = Foo(pk=sql.select(sql.func.coalesce(sql.func.max(Foo.pk) + 1, 1)))
session.add(foo)
session.commit()

INSERT  INTO  foo  (foopk,  bar)  VALUES
((SELECT  coalesce(max(foo.foopk)  +  %(max_1)s,  %(coalesce_2)s)  AS  coalesce_1
FROM  foo),  %(bar)s)  RETURNING  foo.foopk

Session = sessionmaker(bind=engine)
session = Session()
# execute a string statement
result = session.execute(text("select * from table where id=:id"), {"id": 7})
# execute a SQL expression construct
result = session.execute(select(mytable).where(mytable.c.id == 7))

connection = session.connection()

Session = sessionmaker()
session = Session()
# need to specify mapper or class when executing
result = session.execute(
    text("select * from table where id=:id"),
    {"id": 7},
    bind_arguments={"mapper": MyMappedClass},
)
result = session.execute(
    select(mytable).where(mytable.c.id == 7), bind_arguments={"mapper": MyMappedClass}
)
connection = session.connection(MyMappedClass)

class MyObject(Base):
    __tablename__ = "my_table"
    id = mapped_column(Integer, primary_key=True)
    data = mapped_column(String(50), nullable=True)
obj = MyObject(id=1)
session.add(obj)
session.commit()  # INSERT with the 'data' column omitted; the database
# itself will persist this as the NULL value

class MyObject(Base):
    __tablename__ = "my_table"
    id = mapped_column(Integer, primary_key=True)
    data = mapped_column(String(50), nullable=True, server_default="default")
obj = MyObject(id=1)
session.add(obj)
session.commit()  # INSERT with the 'data' column omitted; the database
# itself will persist this as the value 'default'

class MyObject(Base):
    __tablename__ = "my_table"
    id = mapped_column(Integer, primary_key=True)
    data = mapped_column(String(50), nullable=True, server_default="default")
obj = MyObject(id=1, data=None)
session.add(obj)
session.commit()  # INSERT with the 'data' column explicitly set to None;
# the ORM still omits it from the statement and the
# database will still persist this as the value 'default'

from sqlalchemy import null
obj = MyObject(id=1, data=null())
session.add(obj)
session.commit()  # INSERT with the 'data' column explicitly set as null();
# the ORM uses this directly, bypassing all client-
# and server-side defaults, and the database will
# persist this as the NULL value

class MyObject(Base):
    __tablename__ = "my_table"
    id = mapped_column(Integer, primary_key=True)
    data = mapped_column(
        String(50).evaluates_none(),  # indicate that None should always be passed
        nullable=True,
        server_default="default",
    )
obj = MyObject(id=1, data=None)
session.add(obj)
session.commit()  # INSERT with the 'data' column explicitly set to None;
# the ORM uses this directly, bypassing all client-
# and server-side defaults, and the database will
# persist this as the NULL value
```## 获取服务器生成的默认值
如在服务器调用的 DDL-显式默认表达式和标记隐式生成的值、时间戳和触发列章节中介绍的，Core 支持数据库列的概念，即数据库自身在 INSERT 语句中生成一个值，以及在较少见的情况下，在 UPDATE 语句中生成一个值。ORM 功能支持此类列，以便在刷新时能够获取这些新生成的值。在服务器生成的主键列的情况下，这种行为是必需的，因为一旦对象被持久化，ORM 就必须知道对象的主键。
在绝大多数情况下，由数据库自动生成值的主键列是简单的整数列，数据库实现为所谓的“自增”列，或者从与列关联的序列中生成。SQLAlchemy Core 中的每个数据库方言都支持一种检索这些主键值的方法，这种方法通常是 Python DBAPI 本地的，并且一般情况下这个过程是自动的。关于这一点还有更多的文档资料在 `Column.autoincrement`。
对于不是主键列或不是简单自增整数列的服务器生成列，ORM 要求这些列用适当的 `server_default` 指令标记，以允许 ORM 检索此值。然而，并非所有方法都支持所有后端，因此必须小心使用适当的方法。要回答的两个问题是，1\. 这一列是否是主键，2\. 数据库是否支持 RETURNING 或等效方法，如“OUTPUT inserted”；这些是在调用 INSERT 或 UPDATE 语句时同时返回服务器生成的值的 SQL 短语。RETURNING 目前由 PostgreSQL、Oracle、MariaDB 10.5、SQLite 3.35 和 SQL Server 支持。
### 情况 1：非主键，支持 RETURNING 或等效方法
在这种情况下，列应标记为 `FetchedValue` 或具有显式的 `Column.server_default`。当执行 INSERT 语句时，ORM 将自动将这些列添加到 RETURNING 子句中，假设 `Mapper.eager_defaults` 参数设置为 `True`，或者如果将其默认设置为 `"auto"`，对于同时支持 RETURNING 和 insertmanyvalues 的方言：
```py
class MyModel(Base):
    __tablename__ = "my_table"
    id = mapped_column(Integer, primary_key=True)
    # server-side SQL date function generates a new timestamp
    timestamp = mapped_column(DateTime(), server_default=func.now())
    # some other server-side function not named here, such as a trigger,
    # populates a value into this column during INSERT
    special_identifier = mapped_column(String(50), server_default=FetchedValue())
    # set eager defaults to True.  This is usually optional, as if the
    # backend supports RETURNING + insertmanyvalues, eager defaults
    # will take place regardless on INSERT
    __mapper_args__ = {"eager_defaults": True}

INSERT  INTO  my_table  DEFAULT  VALUES  RETURNING  my_table.id,  my_table.timestamp,  my_table.special_identifier

class MyModel(Base):
    __tablename__ = "my_table"
    id: Mapped[int] = mapped_column(primary_key=True)
    data: Mapped[str] = mapped_column(String(50))
    # assume a database trigger populates a value into this column
    # during INSERT
    special_identifier = mapped_column(String(50), server_default=FetchedValue())
    # disable all use of RETURNING for the table
    __table_args__ = {"implicit_returning": False}

INSERT  INTO  my_table  (data)  VALUES  (?);  select  scope_identity()

class MyModel(Base):
    __tablename__ = "my_table"
    id = mapped_column(Integer, primary_key=True)
    timestamp = mapped_column(DateTime(), server_default=func.now())
    # assume a database trigger populates a value into this column
    # during INSERT
    special_identifier = mapped_column(String(50), server_default=FetchedValue())

INSERT  INTO  my_table  ()  VALUES  ()
-- when eager_defaults **is** used, but RETURNING is not supported
SELECT  my_table.timestamp  AS  my_table_timestamp,  my_table.special_identifier  AS  my_table_special_identifier
FROM  my_table  WHERE  my_table.id  =  %s

class MyOracleModel(Base):
    __tablename__ = "my_table"
    id: Mapped[int] = mapped_column(Identity(), primary_key=True)
    data: Mapped[str] = mapped_column(String(50))

INSERT  INTO  my_table  (data)  VALUES  (:data)  RETURNING  my_table.id  INTO  :ret_0

class MyOracleModel(Base):
    __tablename__ = "my_table"
    id: Mapped[int] = mapped_column(Sequence("my_oracle_seq"), primary_key=True)
    data: Mapped[str] = mapped_column(String(50))

INSERT  INTO  my_table  (id,  data)  VALUES  (my_oracle_seq.nextval,  :data)  RETURNING  my_table.id  INTO  :ret_0

class MySQLServerModel(Base):
    __tablename__ = "my_table"
    timestamp: Mapped[datetime.datetime] = mapped_column(
        TIMESTAMP(), server_default=FetchedValue(), primary_key=True
    )
    data: Mapped[str] = mapped_column(String(50))

INSERT  INTO  my_table  (data)  OUTPUT  inserted.timestamp  VALUES  (?)

class MyModel(Base):
    __tablename__ = "my_table"
    timestamp = mapped_column(DateTime(), default=func.now(), primary_key=True)

SELECT  now()  AS  anon_1
INSERT  INTO  my_table  (timestamp)  VALUES  (%s)
('2018-08-09 13:08:46',)

from sqlalchemy import cast, Binary
class MyModel(Base):
    __tablename__ = "my_table"
    timestamp = mapped_column(
        TIMESTAMP(), default=cast(func.now(), Binary), primary_key=True
    )

SELECT  CAST(now()  AS  BINARY)  AS  anon_1
INSERT  INTO  my_table  (timestamp)  VALUES  (%s)
(b'2018-08-09 13:08:46',)

class MyModel(Base):
    __tablename__ = "my_table"
    id = mapped_column(Integer, primary_key=True)
    created = mapped_column(
        DateTime(), default=func.now(), server_default=FetchedValue()
    )
    updated = mapped_column(
        DateTime(),
        onupdate=func.now(),
        server_default=FetchedValue(),
        server_onupdate=FetchedValue(),
    )
    __mapper_args__ = {"eager_defaults": True}

INSERT  INTO  my_table  (created)  VALUES  (now())  RETURNING  my_table.id,  my_table.created,  my_table.updated
UPDATE  my_table  SET  updated=now()  WHERE  my_table.id  =  %(my_table_id)s  RETURNING  my_table.updated
```## 使用 INSERT、UPDATE 和 ON CONFLICT（即 upsert）返回 ORM 对象
SQLAlchemy 2.0 包括增强功能，用于发出几种类型的 ORM-enabled INSERT、UPDATE 和 upsert 语句。请查看文档 ORM-Enabled INSERT, UPDATE, and DELETE statements。有关 upsert，请参见 ORM “upsert” Statements。
### 使用 PostgreSQL ON CONFLICT 与 RETURNING 返回 upserted ORM 对象
本节已移至 ORM "upsert"语句。## 分区策略（例如，每个会话多个数据库后端）
### 简单的垂直分区
垂直分区将不同的类、类层次结构或映射表配置到多个数据库中，方法是通过使用`Session`的`Session.binds`参数。该参数接收一个包含任意 ORM 映射类、映射层次结构中的任意类（如声明基类或混合类）、`Table`对象和`Mapper`对象的字典作为键，然后通常引用`Engine`或不太常见的是`Connection`对象作为目标。每当`Session`需要代表特定类型的映射类发出 SQL 以定位适当的数据库连接源时，都会查询该字典：
```py
engine1 = create_engine("postgresql+psycopg2://db1")
engine2 = create_engine("postgresql+psycopg2://db2")
Session = sessionmaker()
# bind User operations to engine 1, Account operations to engine 2
Session.configure(binds={User: engine1, Account: engine2})
session = Session()

from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Session
class BaseA(DeclarativeBase):
    pass
class BaseB(DeclarativeBase):
    pass
class User(BaseA): ...
class Address(BaseA): ...
class GameInfo(BaseB): ...
class GameStats(BaseB): ...
Session = sessionmaker()
# all User/Address operations will be on engine 1, all
# Game operations will be on engine 2
Session.configure(binds={BaseA: engine1, BaseB: engine2})

engines = {
    "leader": create_engine("sqlite:///leader.db"),
    "other": create_engine("sqlite:///other.db"),
    "follower1": create_engine("sqlite:///follower1.db"),
    "follower2": create_engine("sqlite:///follower2.db"),
}
from sqlalchemy.sql import Update, Delete
from sqlalchemy.orm import Session, sessionmaker
import random
class RoutingSession(Session):
    def get_bind(self, mapper=None, clause=None):
        if mapper and issubclass(mapper.class_, MyOtherClass):
            return engines["other"]
        elif self._flushing or isinstance(clause, (Update, Delete)):
            # NOTE: this is for example, however in practice reader/writer
            # splits are likely more straightforward by using two distinct
            # Sessions at the top of a "reader" or "writer" operation.
            # See note below
            return engines["leader"]
        else:
            return engines[random.choice(["follower1", "follower2"])]

Session = sessionmaker(class_=RoutingSession)

class SomeClass(Base):
    __tablename__ = "some_table"
    # ...
    value = mapped_column(Integer)
someobject = session.get(SomeClass, 5)
# set 'value' attribute to a SQL expression adding one
someobject.value = SomeClass.value + 1
# issues "UPDATE some_table SET value=value+1"
session.commit()

class Foo(Base):
    __tablename__ = "foo"
    pk = mapped_column(Integer, primary_key=True)
    bar = mapped_column(Integer)
e = create_engine("postgresql+psycopg2://scott:tiger@localhost/test", echo=True)
Base.metadata.create_all(e)
session = Session(e)
foo = Foo(pk=sql.select(sql.func.coalesce(sql.func.max(Foo.pk) + 1, 1)))
session.add(foo)
session.commit()

INSERT  INTO  foo  (foopk,  bar)  VALUES
((SELECT  coalesce(max(foo.foopk)  +  %(max_1)s,  %(coalesce_2)s)  AS  coalesce_1
FROM  foo),  %(bar)s)  RETURNING  foo.foopk