SqlAlchemy 2.0 中文文档（五十七）

# (variable) str_col: ColumnClause[str]
str_col = column("a", String)

# (variable) int_col: ColumnClause[int]
int_col = column("a", Integer)

# (variable) expr1: ColumnElement[str]
expr1 = str_col + "x"

# (variable) expr2: ColumnElement[int]
expr2 = int_col + 10

# (variable) expr3: ColumnElement[bool]
expr3 = int_col == 15

# (variable) stmt: Select[Tuple[str, int]]
stmt = select(str_col, int_col)

# (variable) stmt: ReturningInsert[Tuple[str, int]]
ins_stmt = insert(table("t")).returning(str_col, int_col)

with engine.connect() as conn:
    # (variable) stmt: Select[Tuple[str, int]]
    stmt = select(str_col, int_col)

    # (variable) result: Result[Tuple[str, int]]
    result = conn.execute(stmt)

    # (variable) row: Row[Tuple[str, int]] | None
    row = result.first()

    if row is not None:
        # for typed tuple unpacking or indexed access,
        # use row.tuple() or row.t  (this is the small typing-oriented accessor)
        strval, intval = row.t

        # (variable) strval: str
        strval

        # (variable) intval: int
        intval

# (variable) data: Sequence[str]
data = connection.execute(select(str_col)).scalars().all()

from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    addresses: Mapped[List["Address"]] = relationship()

class Address(Base):
    __tablename__ = "address"

    id: Mapped[int] = mapped_column(primary_key=True)
    email_address: Mapped[str]
    user_id = mapped_column(ForeignKey("user_account.id"))

with Session(engine) as session:
    # (variable) stmt: Select[Tuple[int, str]]
    stmt_1 = select(User.id, User.name)

    # (variable) result_1: Result[Tuple[int, str]]
    result_1 = session.execute(stmt_1)

    # (variable) intval: int
    # (variable) strval: str
    intval, strval = result_1.one().t

with Session(engine) as session:
    # (variable) stmt: Select[Tuple[User, Address]]
    stmt_2 = select(User, Address).join_from(User, Address)

    # (variable) result_2: Result[Tuple[User, Address]]
    result_2 = session.execute(stmt_2)

    # (variable) user_obj: User
    # (variable) address_obj: Address
    user_obj, address_obj = result_2.one().t

with Session(engine) as session:
    # this is in fact an Annotated type, but typing tools don't
    # generally display this

    # (variable) u1: Type[User]
    u1 = aliased(User)

    # (variable) stmt: Select[Tuple[User, User, str]]
    stmt = select(User, u1, User.name).filter(User.id == 5)

    # (variable) result: Result[Tuple[User, User, str]]
    result = session.execute(stmt)

# (variable) users1: Sequence[User]
users1 = session.scalars(select(User)).all()

# (variable) user: User
user = session.query(User).one()

# (variable) user_iter: Iterator[User]
user_iter = iter(session.scalars(select(User)))

# (variable) q1: RowReturningQuery[Tuple[int, str]]
q1 = session.query(User.id, User.name)

# (variable) rows: List[Row[Tuple[int, str]]]
rows = q1.all()

# (variable) q2: Query[User]
q2 = session.query(User)

# (variable) users: List[User]
users = q2.all()

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

from sqlalchemy import Column
from sqlalchemy.orm import relationship
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id = Column(Integer, primary_key=True)
    name = Column(String(30), nullable=False)
    fullname = Column(String)
    addresses = relationship("Address", back_populates="user")

class Address(Base):
    __tablename__ = "address"

    id = Column(Integer, primary_key=True)
    email_address = Column(String, nullable=False)
    user_id = Column(ForeignKey("user_account.id"), nullable=False)
    user = relationship("User", back_populates="addresses")

from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id = mapped_column(Integer, primary_key=True)
    name = mapped_column(String(30), nullable=False)
    fullname = mapped_column(String)
    addresses = relationship("Address", back_populates="user")

class Address(Base):
    __tablename__ = "address"

    id = mapped_column(Integer, primary_key=True)
    email_address = mapped_column(String, nullable=False)
    user_id = mapped_column(ForeignKey("user_account.id"), nullable=False)
    user = relationship("User", back_populates="addresses")

from typing import List
from typing import Optional
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(30), nullable=False)
    fullname: Mapped[Optional[str]] = mapped_column(String)
    addresses: Mapped[List["Address"]] = relationship("Address", back_populates="user")

class Address(Base):
    __tablename__ = "address"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    email_address: Mapped[str] = mapped_column(String, nullable=False)
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"), nullable=False)
    user: Mapped["User"] = relationship("User", back_populates="addresses")

from typing import List
from typing import Optional
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    fullname: Mapped[Optional[str]]
    addresses: Mapped[List["Address"]] = relationship(back_populates="user")

class Address(Base):
    __tablename__ = "address"

    id: Mapped[int] = mapped_column(primary_key=True)
    email_address: Mapped[str]
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
    user: Mapped["User"] = relationship(back_populates="addresses")

from typing_extensions import Annotated
from sqlalchemy.orm import DeclarativeBase

str50 = Annotated[str, 50]

# declarative base with a type-level override, using a type that is
# expected to be used in multiple places
class Base(DeclarativeBase):
    type_annotation_map = {
        str50: String(50),
    }

from typing_extensions import Annotated
from typing import List
from typing import Optional
from sqlalchemy import ForeignKey
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

# declarative base from previous example
str50 = Annotated[str, 50]

class Base(DeclarativeBase):
    type_annotation_map = {
        str50: String(50),
    }

# set up mapped_column() overrides, using whole column styles that are
# expected to be used in multiple places
intpk = Annotated[int, mapped_column(primary_key=True)]
user_fk = Annotated[int, mapped_column(ForeignKey("user_account.id"))]

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[intpk]
    name: Mapped[str50]
    fullname: Mapped[Optional[str]]
    addresses: Mapped[List["Address"]] = relationship(back_populates="user")

class Address(Base):
    __tablename__ = "address"

    id: Mapped[intpk]
    email_address: Mapped[str50]
    user_id: Mapped[user_fk]
    user: Mapped["User"] = relationship(back_populates="addresses")

# (variable) stmt: Select[Tuple[int, str]]
stmt = select(User.id, User.name)

with Session(e) as sess:
    for row in sess.execute(stmt):
        # (variable) row: Row[Tuple[int, str]]
        print(row)

    # (variable) users: Sequence[User]
    users = sess.scalars(select(User)).all()

    # (variable) users_legacy: List[User]
    users_legacy = sess.query(User).all()

from typing_extensions import Annotated
from typing import List
from typing import Optional
from sqlalchemy import ForeignKey
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import MappedAsDataclass
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(MappedAsDataclass, DeclarativeBase):
  """subclasses will be converted to dataclasses"""

intpk = Annotated[int, mapped_column(primary_key=True)]
str30 = Annotated[str, mapped_column(String(30))]
user_fk = Annotated[int, mapped_column(ForeignKey("user_account.id"))]

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[intpk] = mapped_column(init=False)
    name: Mapped[str30]
    fullname: Mapped[Optional[str]] = mapped_column(default=None)
    addresses: Mapped[List["Address"]] = relationship(
        back_populates="user", default_factory=list
    )

class Address(Base):
    __tablename__ = "address"

    id: Mapped[intpk] = mapped_column(init=False)
    email_address: Mapped[str]
    user_id: Mapped[user_fk] = mapped_column(init=False)
    user: Mapped["User"] = relationship(back_populates="addresses", default=None)

>>> u1 = User("username", fullname="full name", addresses=[Address("email@address")])
>>> u1
User(id=None, name='username', fullname='full name', addresses=[Address(id=None, email_address='email@address', user_id=None, user=...)])

@Profiler.profile
def test_flush_no_pk(n):
  """INSERT statements via the ORM (batched with RETURNING if available),
 fetching generated row id"""
    session = Session(bind=engine)
    for chunk in range(0, n, 1000):
        session.add_all(
            [
                Customer(
                    name="customer name %d" % i,
                    description="customer description %d" % i,
                )
                for i in range(chunk, chunk + 1000)
            ]
        )
        session.flush()
    session.commit()

python -m examples.performance.bulk_inserts --test test_flush_no_pk

>>> users = session.scalars(
...     insert(User).returning(User),
...     [
...         {"name": "spongebob", "fullname": "Spongebob Squarepants"},
...         {"name": "sandy", "fullname": "Sandy Cheeks"},
...         {"name": "patrick", "fullname": "Patrick Star"},
...         {"name": "squidward", "fullname": "Squidward Tentacles"},
...         {"name": "ehkrabs", "fullname": "Eugene H. Krabs"},
...     ],
... )
>>> print(users.all())
[User(name='spongebob', fullname='Spongebob Squarepants'),
 User(name='sandy', fullname='Sandy Cheeks'),
 User(name='patrick', fullname='Patrick Star'),
 User(name='squidward', fullname='Squidward Tentacles'),
 User(name='ehkrabs', fullname='Eugene H. Krabs')]

>>> from sqlalchemy import update
>>> session.execute(
...     update(User),
...     [
...         {"id": 1, "fullname": "Spongebob Squarepants"},
...         {"id": 3, "fullname": "Patrick Star"},
...     ],
... )

>>> from sqlalchemy.dialects.sqlite import insert as sqlite_upsert
>>> stmt = sqlite_upsert(User).values(
...     [
...         {"name": "spongebob", "fullname": "Spongebob Squarepants"},
...         {"name": "sandy", "fullname": "Sandy Cheeks"},
...         {"name": "patrick", "fullname": "Patrick Star"},
...         {"name": "squidward", "fullname": "Squidward Tentacles"},
...         {"name": "ehkrabs", "fullname": "Eugene H. Krabs"},
...     ]
... )
>>> stmt = stmt.on_conflict_do_update(
...     index_elements=[User.name], set_=dict(fullname=stmt.excluded.fullname)
... )
>>> result = session.scalars(stmt.returning(User))
>>> print(result.all())
[User(name='spongebob', fullname='Spongebob Squarepants'),
User(name='sandy', fullname='Sandy Cheeks'),
User(name='patrick', fullname='Patrick Star'),
User(name='squidward', fullname='Squidward Tentacles'),
User(name='ehkrabs', fullname='Eugene H. Krabs')]

>>> from sqlalchemy import update
>>> stmt = (
...     update(User)
...     .where(User.name == "squidward")
...     .values(name="spongebob")
...     .returning(User)
... )
>>> result = session.scalars(stmt, execution_options={"synchronize_session": "fetch"})
>>> print(result.all())

from sqlalchemy.orm import WriteOnlyMapped

class Base(DeclarativeBase):
    pass

class Account(Base):
    __tablename__ = "account"
    id: Mapped[int] = mapped_column(primary_key=True)
    identifier: Mapped[str]
    account_transactions: WriteOnlyMapped["AccountTransaction"] = relationship(
        cascade="all, delete-orphan",
        passive_deletes=True,
        order_by="AccountTransaction.timestamp",
    )

class AccountTransaction(Base):
    __tablename__ = "account_transaction"
    id: Mapped[int] = mapped_column(primary_key=True)
    account_id: Mapped[int] = mapped_column(
        ForeignKey("account.id", ondelete="cascade")
    )
    description: Mapped[str]
    amount: Mapped[Decimal]
    timestamp: Mapped[datetime] = mapped_column(default=func.now())

new_account = Account(
    identifier="account_01",
    account_transactions=[
        AccountTransaction(description="initial deposit", amount=Decimal("500.00")),
        AccountTransaction(description="transfer", amount=Decimal("1000.00")),
        AccountTransaction(description="withdrawal", amount=Decimal("-29.50")),
    ],
)

new_account.account_transactions.add(
    AccountTransaction(description="transfer", amount=Decimal("2000.00"))
)

account_transactions = session.scalars(
    existing_account.account_transactions.select()
    .where(AccountTransaction.amount < 0)
    .limit(10)
).all()

from sqlalchemy.orm import DynamicMapped

class Base(DeclarativeBase):
    pass

class Account(Base):
    __tablename__ = "account"
    id: Mapped[int] = mapped_column(primary_key=True)
    identifier: Mapped[str]
    account_transactions: DynamicMapped["AccountTransaction"] = relationship(
        cascade="all, delete-orphan",
        passive_deletes=True,
        order_by="AccountTransaction.timestamp",
    )

class AccountTransaction(Base):
    __tablename__ = "account_transaction"
    id: Mapped[int] = mapped_column(primary_key=True)
    account_id: Mapped[int] = mapped_column(
        ForeignKey("account.id", ondelete="cascade")
    )
    description: Mapped[str]
    amount: Mapped[Decimal]
    timestamp: Mapped[datetime] = mapped_column(default=func.now())

meta = MetaData()

my_table = Table(
    "my_table",
    meta,
    Column("id", Integer, primary_key=True),
    Column("num", Integer),
    Column("data", String),
    Index("my_pg_index", "data").ddl_if(dialect="postgresql"),
    CheckConstraint("num > 5").ddl_if(dialect="postgresql"),
)

e1 = create_engine("sqlite://", echo=True)
meta.create_all(e1)  # will not generate CHECK and INDEX

e2 = create_engine("postgresql://scott:tiger@localhost/test", echo=True)
meta.create_all(e2)  # will generate CHECK and INDEX

>>> import datetime

>>> from sqlalchemy import DATETIME
>>> from sqlalchemy import literal
>>> from sqlalchemy.dialects import oracle
>>> from sqlalchemy.dialects import postgresql

>>> date_literal = literal(datetime.datetime.now(), DATETIME)

>>> print(
...     date_literal.compile(
...         dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}
...     )
... )
'2022-12-17 11:02:13.575789'
>>> print(
...     date_literal.compile(
...         dialect=oracle.dialect(), compile_kwargs={"literal_binds": True}
...     )
... )
TO_TIMESTAMP('2022-12-17 11:02:13.575789',  'YYYY-MM-DD HH24:MI:SS.FF') 

with engine.connect() as conn:
    with conn.execution_options(yield_per=100).execute(
        text("select * from table")
    ) as result:
        for row in result:
            print(f"{row}")

async with async_engine.connect() as conn:
    async with conn.execution_options(yield_per=100).execute(
        text("select * from table")
    ) as result:
        for row in result:
            print(f"{row}")

class SomeTest(TestCase):
    def setUp(self):
        # connect to the database
        self.connection = engine.connect()

        # begin a non-ORM transaction
        self.trans = self.connection.begin()

        # bind an individual Session to the connection, selecting
        # "create_savepoint" join_transaction_mode
        self.session = Session(
            bind=self.connection, join_transaction_mode="create_savepoint"
        )

    def tearDown(self):
        self.session.close()

        # rollback non-ORM transaction
        self.trans.rollback()

        # return connection to the Engine
        self.connection.close()

engine = create_engine("...")

# setup outer connection with a transaction and a SAVEPOINT
conn = engine.connect()
trans = conn.begin()
nested = conn.begin_nested()

# bind a Session to that connection and operate upon it, including
# a commit
session = Session(conn)
session.connection()
session.commit()
session.close()

# assert both SAVEPOINT and transaction remain active
assert nested.is_active
nested.rollback()
trans.rollback()

>>> e1 = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")
>>> e2 = create_engine(str(e1.url))

>>> e1 = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")
>>> e2 = create_engine(e1.url)

>>> e1 = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")
>>> url_string = e1.url.render_as_string(hide_password=False)
>>> e2 = create_engine(url_string)

class Foo:
    col1 = mapped_column(Integer)
    col3 = mapped_column(Integer)

class Bar:
    col2 = mapped_column(Integer)
    col4 = mapped_column(Integer)

class Model(Base, Foo, Bar):
    id = mapped_column(Integer, primary_key=True)
    __tablename__ = "model"

CREATE  TABLE  model  (
  col1  INTEGER,
  col3  INTEGER,
  col2  INTEGER,
  col4  INTEGER,
  id  INTEGER  NOT  NULL,
  PRIMARY  KEY  (id)
)

CREATE  TABLE  model  (
  id  INTEGER  NOT  NULL,
  col1  INTEGER,
  col3  INTEGER,
  col2  INTEGER,
  col4  INTEGER,
  PRIMARY  KEY  (id)
)

class Foo:
    id = mapped_column(Integer, primary_key=True)
    col1 = mapped_column(Integer)
    col3 = mapped_column(Integer)

class Model(Foo, Base):
    col2 = mapped_column(Integer)
    col4 = mapped_column(Integer)
    __tablename__ = "model"

CREATE  TABLE  model  (
  col2  INTEGER,
  col4  INTEGER,
  id  INTEGER  NOT  NULL,
  col1  INTEGER,
  col3  INTEGER,
  PRIMARY  KEY  (id)
)

class Foo:
    id = mapped_column(Integer, primary_key=True, sort_order=-10)
    col1 = mapped_column(Integer, sort_order=-1)
    col3 = mapped_column(Integer)

class Model(Foo, Base):
    col2 = mapped_column(Integer)
    col4 = mapped_column(Integer)
    __tablename__ = "model"

CREATE  TABLE  model  (
  id  INTEGER  NOT  NULL,
  col1  INTEGER,
  col2  INTEGER,
  col4  INTEGER,
  col3  INTEGER,
  PRIMARY  KEY  (id)
)

>>> # SQLAlchemy 1.3 (and 2.0)
>>> from sqlalchemy import Sequence
>>> from sqlalchemy.schema import CreateSequence
>>> print(CreateSequence(Sequence("my_seq")))
CREATE  SEQUENCE  my_seq 

>>> # SQLAlchemy 1.4 (only)
>>> from sqlalchemy import Sequence
>>> from sqlalchemy.schema import CreateSequence
>>> print(CreateSequence(Sequence("my_seq")))
CREATE  SEQUENCE  my_seq  START  WITH  1 

>>> # All SQLAlchemy versions
>>> from sqlalchemy import Sequence
>>> from sqlalchemy.schema import CreateSequence
>>> print(CreateSequence(Sequence("my_seq", start=1)))
CREATE  SEQUENCE  my_seq  START  WITH  1 

import typing

from sqlalchemy import String
from sqlalchemy.dialects.mysql import VARCHAR

type_ = String(255).with_variant(VARCHAR(255, charset="utf8mb4"), "mysql", "mariadb")

if typing.TYPE_CHECKING:
    reveal_type(type_)

info: Type of "type_" is "String"

expr = literal(5, Integer) / literal(10, Integer)

%(param_1)s  /  CAST(%(param_2)s  AS  NUMERIC)

expr = literal(5, Integer) // literal(10, Integer)

FLOOR(%(param_1)s  /  %(param_2)s)

expr = func.floor(literal(5, Integer) / literal(10, Integer))

Traceback (most recent call last):
File "/home/classic/dev/sqlalchemy/test3.py", line 30, in worker
    sess.execute(select(A)).all()
File "/home/classic/tmp/sqlalchemy/lib/sqlalchemy/orm/session.py", line 1691, in execute
    conn = self._connection_for_bind(bind)
File "/home/classic/tmp/sqlalchemy/lib/sqlalchemy/orm/session.py", line 1532, in _connection_for_bind
    return self._transaction._connection_for_bind(
File "/home/classic/tmp/sqlalchemy/lib/sqlalchemy/orm/session.py", line 754, in _connection_for_bind
    if self.session.twophase and self._parent is None:
AttributeError: 'NoneType' object has no attribute 'twophase'

File "/home/classic/dev/sqlalchemy/lib/sqlalchemy/orm/session.py", line 1785, in close
   self._close_impl(invalidate=False)
File "/home/classic/dev/sqlalchemy/lib/sqlalchemy/orm/session.py", line 1827, in _close_impl
   transaction.close(invalidate)
File "<string>", line 2, in close
File "/home/classic/dev/sqlalchemy/lib/sqlalchemy/orm/session.py", line 506, in _go
   raise sa_exc.InvalidRequestError(
sqlalchemy.exc.InvalidRequestError: Method 'close()' can't be called here;
method '_connection_for_bind()' is already in progress and this would cause
an unexpected state change to symbol('CLOSED')

from sqlalchemy.dialects import oracle

Table("some_table", metadata, Column("value", oracle.FLOAT(126)))

from sqlalchemy.types import Float
from sqlalchemy.dialects import oracle

Table(
    "some_table",
    metadata,
    Column("value", Float(5).with_variant(oracle.FLOAT(16), "oracle")),
)
```### PostgreSQL 后端的新 RANGE / MULTIRANGE 支持和更改

RANGE / MULTIRANGE 支持已完全实现为 psycopg2、psycopg3 和 asyncpg 方言。新支持使用了一个新的 SQLAlchemy 特定的 `Range` 对象，它对不同的后端是不可知的，不需要使用后端特定的导入或扩展步骤。对于多范围支持，使用 `Range` 对象的列表。

之前使用的特定于 psycopg2 的类型的代码应修改为使用`Range`，它提供了兼容的接口。

`Range` 对象还具有与 PostgreSQL 相同的比较支持。到目前为止已实现了 `Range.contains()` 和 `Range.contained_by()` 方法，它们的工作方式与 PostgreSQL 的 `@>` 和 `<@` 相同。未来的版本可能会添加其他运算符支持。

请参阅范围和多范围类型处的文档，了解如何使用新功能的背景信息。

另请参阅

范围和多范围类型

[#7156](https://www.sqlalchemy.org/trac/ticket/7156) [#8706](https://www.sqlalchemy.org/trac/ticket/8706)  ### 在 PostgreSQL 上使用`match()`操作符使用`plainto_tsquery()`而不是`to_tsquery()`

`Operators.match()`函数现在在 PostgreSQL 后端上呈现为`col @@ plainto_tsquery(expr)`，而不是`col @@ to_tsquery()`。`plainto_tsquery()`接受纯文本，而`to_tsquery()`接受专门的查询符号，因此与其他后端的兼容性较差。

通过使用`func`生成 PostgreSQL 特定函数和`Operators.bool_op()`（`Operators.op()`的布尔类型版本）来生成任意运算符，以与之前版本相同的方式，所有 PostgreSQL 搜索函数和操作符都可用。请参阅全文搜索中的示例。

现有的 SQLAlchemy 项目如果在`Operators.match()`中使用了 PG 特定的指令，应该直接使用`func.to_tsquery()`。要以与 1.4 版本完全相同的形式呈现 SQL，请参阅使用 match()进行简单纯文本匹配的版本说明。

[#7086](https://www.sqlalchemy.org/trac/ticket/7086)

## Core 和 ORM 中的新类型支持-不再使用存根/扩展

与 1.4 版本中通过[sqlalchemy2-stubs](https://github.com/sqlalchemy/sqlalchemy2-stubs)包提供的临时方法相比，Core 和 ORM 的类型化方法已完全重写。新方法始于 SQLAlchemy 中最基本的元素，即`Column`，更准确地说是底层所有具有类型的 SQL 表达式的`ColumnElement`。然后，这个表达式级别的类型化扩展到语句构造、语句执行和结果集的领域，最终扩展到 ORM，其中新的声明性形式允许完全类型化的 ORM 模型，从语句到结果集全部集成。

提示

对于 2.0 系列，类型支持应被视为**beta 级别**软件。类型详细信息可能会发生变化，但不计划进行重大不兼容更改。

### SQL 表达式/语句/结果集类型化

本节提供了 SQLAlchemy 的新 SQL 表达类型方法的背景和示例，它从基本的`ColumnElement`构造扩展到 SQL 语句和结果集，以及 ORM 映射的领域。

#### 原因和概述

提示

本节是一个架构讨论。 跳转到 SQL 表达类型 - 示例以查看新的类型化外观。

在[sqlalchemy2-stubs](https://github.com/sqlalchemy/sqlalchemy2-stubs)中，SQL 表达式被类型化为[泛型](https://peps.python.org/pep-0484/#generics)，然后引用`TypeEngine`对象，如`Integer`，`DateTime`，或`String`作为它们的泛型参数（如`Column[Integer]`）。 这本身就是与最初的 Dropbox [sqlalchemy-stubs](https://github.com/dropbox/sqlalchemy-stubs)包不同的地方，其中`Column`及其基本构造直接以 Python 类型为泛型，例如`int`，`datetime`和`str`。 希望由于`Integer` / `DateTime` / `String`本身是对`int` / `datetime` / `str`的泛型，因此有方法可以保持两个级别的信息，并且可以通过`TypeEngine`从列表达式中提取 Python 类型作为中间构造。 但是，事实并非如此，因为[**PEP 484**](https://peps.python.org/pep-0484/)没有足够丰富的功能集使其可行，缺乏诸如[higher kinded TypeVars](https://github.com/python/typing/issues/548)之类的功能。

经过对[当前能力的深入评估](https://github.com/python/typing/discussions/999)，SQLAlchemy 2.0 实现了[**PEP 484**](https://peps.python.org/pep-0484/)的原始智慧，直接将列表达式与 Python 类型进行关联。这意味着，如果有不同子类型的 SQL 表达式，比如`Column(VARCHAR)`和`Column(Unicode)`，那么这两个`String`子类型的具体信息不会随类型一起传递，因为类型只会携带`str`，但实际上这通常不是问题，通常更有用的是 Python 类型立即出现，因为它代表了将直接存储和接收的 Python 数据。

具体来说，这意味着像`Column('id', Integer)`这样的表达式被类型化为`Column[int]`。这允许建立一个可行的 SQLAlchemy 构造 -> Python 数据类型的管道，而无需使用类型插件。至关重要的是，它允许与 ORM 使用`select()`和`Row`构造的完全互操作性，这些构造引用 ORM 映射的类类型（例如，包含用户映射实例的`Row`，如我们教程中使用的`User`和`Address`示例）。虽然 Python 类型当前对元组类型的定制支持非常有限（其中[**PEP 646**](https://peps.python.org/pep-0646/)是第一个试图处理类似元组对象的 pep，[故意在功能上受到限制](https://mail.python.org/archives/list/typing-sig@python.org/message/G2PNHRR32JMFD3JR7ACA2NDKWTDSEPUG/)，并且本身尚不适用于任意元组操作），但已经设计出了一个相当不错的方法，允许基本的`select()` -> `Result` -> `Row`类型化功能，包括 ORM 类，其中在将`Row`对象解包为单独列条目时，添加了一个小的面向类型的访问器，允许各个 Python 值保持与其来源的 SQL 表达式相关联的 Python 类型（翻译：它有效）。

#### SQL 表达式类型化 - 示例

简要介绍了类型行为。注释指示在[vscode](https://code.visualstudio.com/)中悬停在代码上会看到什么（或者使用[reveal_type()](https://mypy.readthedocs.io/en/latest/common_issues.html?highlight=reveal_type#reveal-type)辅助工具时大致会显示什么）：

+   简单的 Python 类型分配给 SQL 表达式

    ```py
    # (variable) str_col: ColumnClause[str]
    str_col = column("a", String)

    # (variable) int_col: ColumnClause[int]
    int_col = column("a", Integer)

    # (variable) expr1: ColumnElement[str]
    expr1 = str_col + "x"

    # (variable) expr2: ColumnElement[int]
    expr2 = int_col + 10

    # (variable) expr3: ColumnElement[bool]
    expr3 = int_col == 15
    ```

+   分配给`select()`构造的单个 SQL 表达式以及任何返回行的构造，包括返回行的 DML，如带有`Insert.returning()`的`Insert`，都打包成一个保留每个元素 Python 类型的`Tuple[]`类型。

    ```py
    # (variable) stmt: Select[Tuple[str, int]]
    stmt = select(str_col, int_col)

    # (variable) stmt: ReturningInsert[Tuple[str, int]]
    ins_stmt = insert(table("t")).returning(str_col, int_col)
    ```

+   任何返回行的结构中的`Tuple[]`类型，在调用`.execute()`方法时，传递到`Result`和`Row`。为了将`Row`对象解包为元组，`Row.tuple()`或`Row.t`访问器本质上将`Row`转换为相应的`Tuple[]`（但在运行时仍保持相同的`Row`对象）。

    ```py
    with engine.connect() as conn:
        # (variable) stmt: Select[Tuple[str, int]]
        stmt = select(str_col, int_col)

        # (variable) result: Result[Tuple[str, int]]
        result = conn.execute(stmt)

        # (variable) row: Row[Tuple[str, int]] | None
        row = result.first()

        if row is not None:
            # for typed tuple unpacking or indexed access,
            # use row.tuple() or row.t  (this is the small typing-oriented accessor)
            strval, intval = row.t

            # (variable) strval: str
            strval

            # (variable) intval: int
            intval
    ```

+   对于单列语句的标量值，使用`Connection.scalar()`、`Result.scalars()`等方法是正确的。

    ```py
    # (variable) data: Sequence[str]
    data = connection.execute(select(str_col)).scalars().all()
    ```

+   对于 ORM 映射类，上述对返回行构造的支持与其最配套，因为映射类可以列出其成员的特定类型。下面的示例设置了一个使用新的类型感知语法的类，将在下一节中描述：

    ```py
    from sqlalchemy.orm import DeclarativeBase
    from sqlalchemy.orm import Mapped
    from sqlalchemy.orm import mapped_column

    class Base(DeclarativeBase):
        pass

    class User(Base):
        __tablename__ = "user_account"

        id: Mapped[int] = mapped_column(primary_key=True)
        name: Mapped[str]
        addresses: Mapped[List["Address"]] = relationship()

    class Address(Base):
        __tablename__ = "address"

        id: Mapped[int] = mapped_column(primary_key=True)
        email_address: Mapped[str]
        user_id = mapped_column(ForeignKey("user_account.id"))
    ```

    通过上述映射，属性从语句到结果集一路类型化并表达自己：

    ```py
    with Session(engine) as session:
        # (variable) stmt: Select[Tuple[int, str]]
        stmt_1 = select(User.id, User.name)

        # (variable) result_1: Result[Tuple[int, str]]
        result_1 = session.execute(stmt_1)

        # (variable) intval: int
        # (variable) strval: str
        intval, strval = result_1.one().t
    ```

    映射类本身也是类型，并且行为相同，例如针对两个映射类进行 SELECT 查询：

    ```py
    with Session(engine) as session:
        # (variable) stmt: Select[Tuple[User, Address]]
        stmt_2 = select(User, Address).join_from(User, Address)

        # (variable) result_2: Result[Tuple[User, Address]]
        result_2 = session.execute(stmt_2)

        # (variable) user_obj: User
        # (variable) address_obj: Address
        user_obj, address_obj = result_2.one().t
    ```

    当选择映射类时，像`aliased`这样的构造也能正常工作，保持原始映射类的列级属性以及语句期望的返回类型：

    ```py
    with Session(engine) as session:
        # this is in fact an Annotated type, but typing tools don't
        # generally display this

        # (variable) u1: Type[User]
        u1 = aliased(User)

        # (variable) stmt: Select[Tuple[User, User, str]]
        stmt = select(User, u1, User.name).filter(User.id == 5)

        # (variable) result: Result[Tuple[User, User, str]]
        result = session.execute(stmt)
    ```

+   核心表（Core Table）目前还没有一个合适的方法来在通过`Table.c`访问时维护`Column`对象的类型。

    由于`Table`被设置为类的实例，并且`Table.c`访问器通常通过名称动态访问`Column`对象，因此尚未为此建立类型化方法；需要一些替代语法。

+   ORM 类、标量等工作得很好。

    选择 ORM 类作为标量或元组的典型用例都有效，无论是 2.0 还是 1.x 风格的查询，都可以获得确切的类型，无论是单独还是包含在适当的容器中，如`Sequence[]`、`List[]`或`Iterator[]`：

    ```py
    # (variable) users1: Sequence[User]
    users1 = session.scalars(select(User)).all()

    # (variable) user: User
    user = session.query(User).one()

    # (variable) user_iter: Iterator[User]
    user_iter = iter(session.scalars(select(User)))
    ```

+   传统的`Query`也获得了元组类型。

    对于`Query`的类型支持远远超出了[sqlalchemy-stubs](https://github.com/dropbox/sqlalchemy-stubs)或[sqlalchemy2-stubs](https://github.com/sqlalchemy/sqlalchemy2-stubs)提供的范围，其中标量对象以及元组类型的`Query`对象将保留大多数情况下的结果级别类型：

    ```py
    # (variable) q1: RowReturningQuery[Tuple[int, str]]
    q1 = session.query(User.id, User.name)

    # (variable) rows: List[Row[Tuple[int, str]]]
    rows = q1.all()

    # (variable) q2: Query[User]
    q2 = session.query(User)

    # (variable) users: List[User]
    users = q2.all()
    ```

#### 注意 - 所有存根必须被卸载

类型支持的一个关键警告是**必须卸载所有 SQLAlchemy 存根包**才能使类型化工作。在针对 Python 虚拟环境运行[mypy](https://mypy.readthedocs.io/en/stable/)时，只需卸载这些包。但是，SQLAlchemy 存根包目前也是[typeshed](https://github.com/python/typeshed)的一部分，它本身捆绑在一些类型工具中，如[Pylance](https://github.com/microsoft/pylance-release)，因此在某些情况下可能需要定位这些包的文件并删除它们，如果它们实际上干扰了新的类型化正确工作。

一旦 SQLAlchemy 2.0 以最终状态发布，typeshed 将从其自己的存根源中删除 SQLAlchemy。

### ORM 声明模型

SQLAlchemy 1.4 引入了第一个使用[sqlalchemy2-stubs](https://github.com/sqlalchemy/sqlalchemy2-stubs)和 Mypy Plugin 组合的 SQLAlchemy 本机 ORM 类型支持。在 SQLAlchemy 2.0 中，Mypy 插件**仍然可用，并已更新以与 SQLAlchemy 2.0 的类型系统一起使用**。但是，现在应该将其视为**已弃用**，因为应用程序现在有一条直接的路径来采用新的类型支持，而不使用插件或存根。

#### 概述

新系统的基本方法是，当使用完全声明式模型（即不使用混合声明式或命令式配置，这些配置不变）时，映射列声明首先在运行时通过检查每个属性声明左侧的类型注释来推导，如果存在的话。左手类型注释应该包含在`Mapped`泛型类型中，否则该属性不被视为映射属性。然后属性声明可以引用右侧的`mapped_column()`构造，用于提供有关要生成和映射的`Column`的附加核心级模式信息。如果左侧存在`Mapped`注释，则此右侧声明是可选的；如果左侧没有注释，则`mapped_column()`可以用作`Column`指令的精确替代，其中它将提供更准确（但不精确）的属性类型行为，即使没有注释存在。

这种方法受到 Python [dataclasses](https://docs.python.org/3/library/dataclasses.html)方法的启发，它从左边开始注释，然后允许右边的可选`dataclasses.field()`规范；与 dataclasses 方法的关键区别在于 SQLAlchemy 的方法是严格的**选择加入**，其中使用`Column`的现有映射如果没有任何类型注释，将继续像以往一样工作，并且`mapped_column()`构造可以直接替换`Column`而不需要任何显式类型注释。只有在确切的属性级 Python 类型存在时，才需要使用带有`Mapped`的显式注释。这些注释可以根据需要，按属性基础使用，对于那些特定类型有帮助的属性；使用`mapped_column()`的未注释属性将在实例级别被标记为`Any`。

#### 迁移现有映射

迁移到新的 ORM 方法开始时更加冗长，但随着可用的新功能的充分利用，变得比以前更简洁。以下步骤详细说明了典型的过渡，然后继续说明了一些更多的选项。

##### 第一步 - `declarative_base()`被`DeclarativeBase`取代。

Python 类型中观察到的一个限制是似乎没有能力从函数动态生成一个类，然后被类型工具理解为新类的基础。为了解决这个问题而不使用插件，通常对`declarative_base()`的调用可以替换为使用`DeclarativeBase`类，它产生与通常相同的`Base`对象，只是类型工具理解它：

```py
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

from sqlalchemy import Column
from sqlalchemy.orm import relationship
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id = Column(Integer, primary_key=True)
    name = Column(String(30), nullable=False)
    fullname = Column(String)
    addresses = relationship("Address", back_populates="user")

class Address(Base):
    __tablename__ = "address"

    id = Column(Integer, primary_key=True)
    email_address = Column(String, nullable=False)
    user_id = Column(ForeignKey("user_account.id"), nullable=False)
    user = relationship("User", back_populates="addresses")

from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id = mapped_column(Integer, primary_key=True)
    name = mapped_column(String(30), nullable=False)
    fullname = mapped_column(String)
    addresses = relationship("Address", back_populates="user")

class Address(Base):
    __tablename__ = "address"

    id = mapped_column(Integer, primary_key=True)
    email_address = mapped_column(String, nullable=False)
    user_id = mapped_column(ForeignKey("user_account.id"), nullable=False)
    user = relationship("User", back_populates="addresses")

from typing import List
from typing import Optional
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(30), nullable=False)
    fullname: Mapped[Optional[str]] = mapped_column(String)
    addresses: Mapped[List["Address"]] = relationship("Address", back_populates="user")

class Address(Base):
    __tablename__ = "address"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    email_address: Mapped[str] = mapped_column(String, nullable=False)
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"), nullable=False)
    user: Mapped["User"] = relationship("User", back_populates="addresses")

from typing import List
from typing import Optional
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    fullname: Mapped[Optional[str]]
    addresses: Mapped[List["Address"]] = relationship(back_populates="user")

class Address(Base):
    __tablename__ = "address"

    id: Mapped[int] = mapped_column(primary_key=True)
    email_address: Mapped[str]
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
    user: Mapped["User"] = relationship(back_populates="addresses")

from typing_extensions import Annotated
from sqlalchemy.orm import DeclarativeBase

str50 = Annotated[str, 50]

# declarative base with a type-level override, using a type that is
# expected to be used in multiple places
class Base(DeclarativeBase):
    type_annotation_map = {
        str50: String(50),
    }

from typing_extensions import Annotated
from typing import List
from typing import Optional
from sqlalchemy import ForeignKey
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

# declarative base from previous example
str50 = Annotated[str, 50]

class Base(DeclarativeBase):
    type_annotation_map = {
        str50: String(50),
    }

# set up mapped_column() overrides, using whole column styles that are
# expected to be used in multiple places
intpk = Annotated[int, mapped_column(primary_key=True)]
user_fk = Annotated[int, mapped_column(ForeignKey("user_account.id"))]

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[intpk]
    name: Mapped[str50]
    fullname: Mapped[Optional[str]]
    addresses: Mapped[List["Address"]] = relationship(back_populates="user")

class Address(Base):
    __tablename__ = "address"

    id: Mapped[intpk]
    email_address: Mapped[str50]
    user_id: Mapped[user_fk]
    user: Mapped["User"] = relationship(back_populates="addresses")

# (variable) stmt: Select[Tuple[int, str]]
stmt = select(User.id, User.name)

with Session(e) as sess:
    for row in sess.execute(stmt):
        # (variable) row: Row[Tuple[int, str]]
        print(row)

    # (variable) users: Sequence[User]
    users = sess.scalars(select(User)).all()

    # (variable) users_legacy: List[User]
    users_legacy = sess.query(User).all()

from typing_extensions import Annotated
from typing import List
from typing import Optional
from sqlalchemy import ForeignKey
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import MappedAsDataclass
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(MappedAsDataclass, DeclarativeBase):
  """subclasses will be converted to dataclasses"""

intpk = Annotated[int, mapped_column(primary_key=True)]
str30 = Annotated[str, mapped_column(String(30))]
user_fk = Annotated[int, mapped_column(ForeignKey("user_account.id"))]

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[intpk] = mapped_column(init=False)
    name: Mapped[str30]
    fullname: Mapped[Optional[str]] = mapped_column(default=None)
    addresses: Mapped[List["Address"]] = relationship(
        back_populates="user", default_factory=list
    )

class Address(Base):
    __tablename__ = "address"

    id: Mapped[intpk] = mapped_column(init=False)
    email_address: Mapped[str]
    user_id: Mapped[user_fk] = mapped_column(init=False)
    user: Mapped["User"] = relationship(back_populates="addresses", default=None)

>>> u1 = User("username", fullname="full name", addresses=[Address("email@address")])
>>> u1
User(id=None, name='username', fullname='full name', addresses=[Address(id=None, email_address='email@address', user_id=None, user=...)])

# (variable) str_col: ColumnClause[str]
str_col = column("a", String)

# (variable) int_col: ColumnClause[int]
int_col = column("a", Integer)

# (variable) expr1: ColumnElement[str]
expr1 = str_col + "x"

# (variable) expr2: ColumnElement[int]
expr2 = int_col + 10

# (variable) expr3: ColumnElement[bool]
expr3 = int_col == 15

# (variable) stmt: Select[Tuple[str, int]]
stmt = select(str_col, int_col)

# (variable) stmt: ReturningInsert[Tuple[str, int]]
ins_stmt = insert(table("t")).returning(str_col, int_col)

with engine.connect() as conn:
    # (variable) stmt: Select[Tuple[str, int]]
    stmt = select(str_col, int_col)

    # (variable) result: Result[Tuple[str, int]]
    result = conn.execute(stmt)

    # (variable) row: Row[Tuple[str, int]] | None
    row = result.first()

    if row is not None:
        # for typed tuple unpacking or indexed access,
        # use row.tuple() or row.t  (this is the small typing-oriented accessor)
        strval, intval = row.t

        # (variable) strval: str
        strval

        # (variable) intval: int
        intval

# (variable) data: Sequence[str]
data = connection.execute(select(str_col)).scalars().all()

from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    addresses: Mapped[List["Address"]] = relationship()

class Address(Base):
    __tablename__ = "address"

    id: Mapped[int] = mapped_column(primary_key=True)
    email_address: Mapped[str]
    user_id = mapped_column(ForeignKey("user_account.id"))

with Session(engine) as session:
    # (variable) stmt: Select[Tuple[int, str]]
    stmt_1 = select(User.id, User.name)

    # (variable) result_1: Result[Tuple[int, str]]
    result_1 = session.execute(stmt_1)

    # (variable) intval: int
    # (variable) strval: str
    intval, strval = result_1.one().t

with Session(engine) as session:
    # (variable) stmt: Select[Tuple[User, Address]]
    stmt_2 = select(User, Address).join_from(User, Address)

    # (variable) result_2: Result[Tuple[User, Address]]
    result_2 = session.execute(stmt_2)

    # (variable) user_obj: User
    # (variable) address_obj: Address
    user_obj, address_obj = result_2.one().t

with Session(engine) as session:
    # this is in fact an Annotated type, but typing tools don't
    # generally display this

    # (variable) u1: Type[User]
    u1 = aliased(User)

    # (variable) stmt: Select[Tuple[User, User, str]]
    stmt = select(User, u1, User.name).filter(User.id == 5)

    # (variable) result: Result[Tuple[User, User, str]]
    result = session.execute(stmt)

# (variable) users1: Sequence[User]
users1 = session.scalars(select(User)).all()

# (variable) user: User
user = session.query(User).one()

# (variable) user_iter: Iterator[User]
user_iter = iter(session.scalars(select(User)))

# (variable) q1: RowReturningQuery[Tuple[int, str]]
q1 = session.query(User.id, User.name)

# (variable) rows: List[Row[Tuple[int, str]]]
rows = q1.all()

# (variable) q2: Query[User]
q2 = session.query(User)

# (variable) users: List[User]
users = q2.all()

# (variable) str_col: ColumnClause[str]
str_col = column("a", String)

# (variable) int_col: ColumnClause[int]
int_col = column("a", Integer)

# (variable) expr1: ColumnElement[str]
expr1 = str_col + "x"

# (variable) expr2: ColumnElement[int]
expr2 = int_col + 10

# (variable) expr3: ColumnElement[bool]
expr3 = int_col == 15

# (variable) stmt: Select[Tuple[str, int]]
stmt = select(str_col, int_col)

# (variable) stmt: ReturningInsert[Tuple[str, int]]
ins_stmt = insert(table("t")).returning(str_col, int_col)

with engine.connect() as conn:
    # (variable) stmt: Select[Tuple[str, int]]
    stmt = select(str_col, int_col)

    # (variable) result: Result[Tuple[str, int]]
    result = conn.execute(stmt)

    # (variable) row: Row[Tuple[str, int]] | None
    row = result.first()

    if row is not None:
        # for typed tuple unpacking or indexed access,
        # use row.tuple() or row.t  (this is the small typing-oriented accessor)
        strval, intval = row.t

        # (variable) strval: str
        strval

        # (variable) intval: int
        intval

# (variable) data: Sequence[str]
data = connection.execute(select(str_col)).scalars().all()

from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    addresses: Mapped[List["Address"]] = relationship()

class Address(Base):
    __tablename__ = "address"

    id: Mapped[int] = mapped_column(primary_key=True)
    email_address: Mapped[str]
    user_id = mapped_column(ForeignKey("user_account.id"))

with Session(engine) as session:
    # (variable) stmt: Select[Tuple[int, str]]
    stmt_1 = select(User.id, User.name)

    # (variable) result_1: Result[Tuple[int, str]]
    result_1 = session.execute(stmt_1)

    # (variable) intval: int
    # (variable) strval: str
    intval, strval = result_1.one().t

with Session(engine) as session:
    # (variable) stmt: Select[Tuple[User, Address]]
    stmt_2 = select(User, Address).join_from(User, Address)

    # (variable) result_2: Result[Tuple[User, Address]]
    result_2 = session.execute(stmt_2)

    # (variable) user_obj: User
    # (variable) address_obj: Address
    user_obj, address_obj = result_2.one().t

with Session(engine) as session:
    # this is in fact an Annotated type, but typing tools don't
    # generally display this

    # (variable) u1: Type[User]
    u1 = aliased(User)

    # (variable) stmt: Select[Tuple[User, User, str]]
    stmt = select(User, u1, User.name).filter(User.id == 5)

    # (variable) result: Result[Tuple[User, User, str]]
    result = session.execute(stmt)

# (variable) users1: Sequence[User]
users1 = session.scalars(select(User)).all()

# (variable) user: User
user = session.query(User).one()

# (variable) user_iter: Iterator[User]
user_iter = iter(session.scalars(select(User)))

# (variable) q1: RowReturningQuery[Tuple[int, str]]
q1 = session.query(User.id, User.name)

# (variable) rows: List[Row[Tuple[int, str]]]
rows = q1.all()

# (variable) q2: Query[User]
q2 = session.query(User)

# (variable) users: List[User]
users = q2.all()

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

from sqlalchemy import Column
from sqlalchemy.orm import relationship
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id = Column(Integer, primary_key=True)
    name = Column(String(30), nullable=False)
    fullname = Column(String)
    addresses = relationship("Address", back_populates="user")

class Address(Base):
    __tablename__ = "address"

    id = Column(Integer, primary_key=True)
    email_address = Column(String, nullable=False)
    user_id = Column(ForeignKey("user_account.id"), nullable=False)
    user = relationship("User", back_populates="addresses")

from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id = mapped_column(Integer, primary_key=True)
    name = mapped_column(String(30), nullable=False)
    fullname = mapped_column(String)
    addresses = relationship("Address", back_populates="user")

class Address(Base):
    __tablename__ = "address"

    id = mapped_column(Integer, primary_key=True)
    email_address = mapped_column(String, nullable=False)
    user_id = mapped_column(ForeignKey("user_account.id"), nullable=False)
    user = relationship("User", back_populates="addresses")

from typing import List
from typing import Optional
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(30), nullable=False)
    fullname: Mapped[Optional[str]] = mapped_column(String)
    addresses: Mapped[List["Address"]] = relationship("Address", back_populates="user")

class Address(Base):
    __tablename__ = "address"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    email_address: Mapped[str] = mapped_column(String, nullable=False)
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"), nullable=False)
    user: Mapped["User"] = relationship("User", back_populates="addresses")

from typing import List
from typing import Optional
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    fullname: Mapped[Optional[str]]
    addresses: Mapped[List["Address"]] = relationship(back_populates="user")

class Address(Base):
    __tablename__ = "address"

    id: Mapped[int] = mapped_column(primary_key=True)
    email_address: Mapped[str]
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
    user: Mapped["User"] = relationship(back_populates="addresses")

from typing_extensions import Annotated
from sqlalchemy.orm import DeclarativeBase

str50 = Annotated[str, 50]

# declarative base with a type-level override, using a type that is
# expected to be used in multiple places
class Base(DeclarativeBase):
    type_annotation_map = {
        str50: String(50),
    }

from typing_extensions import Annotated
from typing import List
from typing import Optional
from sqlalchemy import ForeignKey
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

# declarative base from previous example
str50 = Annotated[str, 50]

class Base(DeclarativeBase):
    type_annotation_map = {
        str50: String(50),
    }

# set up mapped_column() overrides, using whole column styles that are
# expected to be used in multiple places
intpk = Annotated[int, mapped_column(primary_key=True)]
user_fk = Annotated[int, mapped_column(ForeignKey("user_account.id"))]

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[intpk]
    name: Mapped[str50]
    fullname: Mapped[Optional[str]]
    addresses: Mapped[List["Address"]] = relationship(back_populates="user")

class Address(Base):
    __tablename__ = "address"

    id: Mapped[intpk]
    email_address: Mapped[str50]
    user_id: Mapped[user_fk]
    user: Mapped["User"] = relationship(back_populates="addresses")

# (variable) stmt: Select[Tuple[int, str]]
stmt = select(User.id, User.name)

with Session(e) as sess:
    for row in sess.execute(stmt):
        # (variable) row: Row[Tuple[int, str]]
        print(row)

    # (variable) users: Sequence[User]
    users = sess.scalars(select(User)).all()

    # (variable) users_legacy: List[User]
    users_legacy = sess.query(User).all()

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

from sqlalchemy import Column
from sqlalchemy.orm import relationship
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id = Column(Integer, primary_key=True)
    name = Column(String(30), nullable=False)
    fullname = Column(String)
    addresses = relationship("Address", back_populates="user")

class Address(Base):
    __tablename__ = "address"

    id = Column(Integer, primary_key=True)
    email_address = Column(String, nullable=False)
    user_id = Column(ForeignKey("user_account.id"), nullable=False)
    user = relationship("User", back_populates="addresses")

from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id = mapped_column(Integer, primary_key=True)
    name = mapped_column(String(30), nullable=False)
    fullname = mapped_column(String)
    addresses = relationship("Address", back_populates="user")

class Address(Base):
    __tablename__ = "address"

    id = mapped_column(Integer, primary_key=True)
    email_address = mapped_column(String, nullable=False)
    user_id = mapped_column(ForeignKey("user_account.id"), nullable=False)
    user = relationship("User", back_populates="addresses")

from typing import List
from typing import Optional
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(30), nullable=False)
    fullname: Mapped[Optional[str]] = mapped_column(String)
    addresses: Mapped[List["Address"]] = relationship("Address", back_populates="user")

class Address(Base):
    __tablename__ = "address"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    email_address: Mapped[str] = mapped_column(String, nullable=False)
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"), nullable=False)
    user: Mapped["User"] = relationship("User", back_populates="addresses")

from typing import List
from typing import Optional
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    fullname: Mapped[Optional[str]]
    addresses: Mapped[List["Address"]] = relationship(back_populates="user")

class Address(Base):
    __tablename__ = "address"

    id: Mapped[int] = mapped_column(primary_key=True)
    email_address: Mapped[str]
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
    user: Mapped["User"] = relationship(back_populates="addresses")

from typing_extensions import Annotated
from sqlalchemy.orm import DeclarativeBase

str50 = Annotated[str, 50]

# declarative base with a type-level override, using a type that is
# expected to be used in multiple places
class Base(DeclarativeBase):
    type_annotation_map = {
        str50: String(50),
    }

from typing_extensions import Annotated
from typing import List
from typing import Optional
from sqlalchemy import ForeignKey
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

# declarative base from previous example
str50 = Annotated[str, 50]

class Base(DeclarativeBase):
    type_annotation_map = {
        str50: String(50),
    }

# set up mapped_column() overrides, using whole column styles that are
# expected to be used in multiple places
intpk = Annotated[int, mapped_column(primary_key=True)]
user_fk = Annotated[int, mapped_column(ForeignKey("user_account.id"))]

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[intpk]
    name: Mapped[str50]
    fullname: Mapped[Optional[str]]
    addresses: Mapped[List["Address"]] = relationship(back_populates="user")

class Address(Base):
    __tablename__ = "address"

    id: Mapped[intpk]
    email_address: Mapped[str50]
    user_id: Mapped[user_fk]
    user: Mapped["User"] = relationship(back_populates="addresses")

# (variable) stmt: Select[Tuple[int, str]]
stmt = select(User.id, User.name)

with Session(e) as sess:
    for row in sess.execute(stmt):
        # (variable) row: Row[Tuple[int, str]]
        print(row)

    # (variable) users: Sequence[User]
    users = sess.scalars(select(User)).all()

    # (variable) users_legacy: List[User]
    users_legacy = sess.query(User).all()

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

from sqlalchemy import Column
from sqlalchemy.orm import relationship
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id = Column(Integer, primary_key=True)
    name = Column(String(30), nullable=False)
    fullname = Column(String)
    addresses = relationship("Address", back_populates="user")

class Address(Base):
    __tablename__ = "address"

    id = Column(Integer, primary_key=True)
    email_address = Column(String, nullable=False)
    user_id = Column(ForeignKey("user_account.id"), nullable=False)
    user = relationship("User", back_populates="addresses")

from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id = mapped_column(Integer, primary_key=True)
    name = mapped_column(String(30), nullable=False)
    fullname = mapped_column(String)
    addresses = relationship("Address", back_populates="user")

class Address(Base):
    __tablename__ = "address"

    id = mapped_column(Integer, primary_key=True)
    email_address = mapped_column(String, nullable=False)
    user_id = mapped_column(ForeignKey("user_account.id"), nullable=False)
    user = relationship("User", back_populates="addresses")

from typing import List
from typing import Optional
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(30), nullable=False)
    fullname: Mapped[Optional[str]] = mapped_column(String)
    addresses: Mapped[List["Address"]] = relationship("Address", back_populates="user")

class Address(Base):
    __tablename__ = "address"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    email_address: Mapped[str] = mapped_column(String, nullable=False)
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"), nullable=False)
    user: Mapped["User"] = relationship("User", back_populates="addresses")

from typing import List
from typing import Optional
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    fullname: Mapped[Optional[str]]
    addresses: Mapped[List["Address"]] = relationship(back_populates="user")

class Address(Base):
    __tablename__ = "address"

    id: Mapped[int] = mapped_column(primary_key=True)
    email_address: Mapped[str]
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
    user: Mapped["User"] = relationship(back_populates="addresses")

from typing_extensions import Annotated
from sqlalchemy.orm import DeclarativeBase

str50 = Annotated[str, 50]

# declarative base with a type-level override, using a type that is
# expected to be used in multiple places
class Base(DeclarativeBase):
    type_annotation_map = {
        str50: String(50),
    }

from typing_extensions import Annotated
from typing import List
from typing import Optional
from sqlalchemy import ForeignKey
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

# declarative base from previous example
str50 = Annotated[str, 50]

class Base(DeclarativeBase):
    type_annotation_map = {
        str50: String(50),
    }

# set up mapped_column() overrides, using whole column styles that are
# expected to be used in multiple places
intpk = Annotated[int, mapped_column(primary_key=True)]
user_fk = Annotated[int, mapped_column(ForeignKey("user_account.id"))]

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[intpk]
    name: Mapped[str50]
    fullname: Mapped[Optional[str]]
    addresses: Mapped[List["Address"]] = relationship(back_populates="user")

class Address(Base):
    __tablename__ = "address"

    id: Mapped[intpk]
    email_address: Mapped[str50]
    user_id: Mapped[user_fk]
    user: Mapped["User"] = relationship(back_populates="addresses")

# (variable) stmt: Select[Tuple[int, str]]
stmt = select(User.id, User.name)

with Session(e) as sess:
    for row in sess.execute(stmt):
        # (variable) row: Row[Tuple[int, str]]
        print(row)

    # (variable) users: Sequence[User]
    users = sess.scalars(select(User)).all()

    # (variable) users_legacy: List[User]
    users_legacy = sess.query(User).all()

from typing_extensions import Annotated
from typing import List
from typing import Optional
from sqlalchemy import ForeignKey
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import MappedAsDataclass
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(MappedAsDataclass, DeclarativeBase):
  """subclasses will be converted to dataclasses"""

intpk = Annotated[int, mapped_column(primary_key=True)]
str30 = Annotated[str, mapped_column(String(30))]
user_fk = Annotated[int, mapped_column(ForeignKey("user_account.id"))]

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[intpk] = mapped_column(init=False)
    name: Mapped[str30]
    fullname: Mapped[Optional[str]] = mapped_column(default=None)
    addresses: Mapped[List["Address"]] = relationship(
        back_populates="user", default_factory=list
    )

class Address(Base):
    __tablename__ = "address"

    id: Mapped[intpk] = mapped_column(init=False)
    email_address: Mapped[str]
    user_id: Mapped[user_fk] = mapped_column(init=False)
    user: Mapped["User"] = relationship(back_populates="addresses", default=None)

>>> u1 = User("username", fullname="full name", addresses=[Address("email@address")])
>>> u1
User(id=None, name='username', fullname='full name', addresses=[Address(id=None, email_address='email@address', user_id=None, user=...)])

@Profiler.profile
def test_flush_no_pk(n):
  """INSERT statements via the ORM (batched with RETURNING if available),
 fetching generated row id"""
    session = Session(bind=engine)
    for chunk in range(0, n, 1000):
        session.add_all(
            [
                Customer(
                    name="customer name %d" % i,
                    description="customer description %d" % i,
                )
                for i in range(chunk, chunk + 1000)
            ]
        )
        session.flush()
    session.commit()

python -m examples.performance.bulk_inserts --test test_flush_no_pk

@Profiler.profile
def test_flush_no_pk(n):
  """INSERT statements via the ORM (batched with RETURNING if available),
 fetching generated row id"""
    session = Session(bind=engine)
    for chunk in range(0, n, 1000):
        session.add_all(
            [
                Customer(
                    name="customer name %d" % i,
                    description="customer description %d" % i,
                )
                for i in range(chunk, chunk + 1000)
            ]
        )
        session.flush()
    session.commit()

python -m examples.performance.bulk_inserts --test test_flush_no_pk

>>> users = session.scalars(
...     insert(User).returning(User),
...     [
...         {"name": "spongebob", "fullname": "Spongebob Squarepants"},
...         {"name": "sandy", "fullname": "Sandy Cheeks"},
...         {"name": "patrick", "fullname": "Patrick Star"},
...         {"name": "squidward", "fullname": "Squidward Tentacles"},
...         {"name": "ehkrabs", "fullname": "Eugene H. Krabs"},
...     ],
... )
>>> print(users.all())
[User(name='spongebob', fullname='Spongebob Squarepants'),
 User(name='sandy', fullname='Sandy Cheeks'),
 User(name='patrick', fullname='Patrick Star'),
 User(name='squidward', fullname='Squidward Tentacles'),
 User(name='ehkrabs', fullname='Eugene H. Krabs')]

>>> from sqlalchemy import update
>>> session.execute(
...     update(User),
...     [
...         {"id": 1, "fullname": "Spongebob Squarepants"},
...         {"id": 3, "fullname": "Patrick Star"},
...     ],
... )

>>> from sqlalchemy.dialects.sqlite import insert as sqlite_upsert
>>> stmt = sqlite_upsert(User).values(
...     [
...         {"name": "spongebob", "fullname": "Spongebob Squarepants"},
...         {"name": "sandy", "fullname": "Sandy Cheeks"},
...         {"name": "patrick", "fullname": "Patrick Star"},
...         {"name": "squidward", "fullname": "Squidward Tentacles"},
...         {"name": "ehkrabs", "fullname": "Eugene H. Krabs"},
...     ]
... )
>>> stmt = stmt.on_conflict_do_update(
...     index_elements=[User.name], set_=dict(fullname=stmt.excluded.fullname)
... )
>>> result = session.scalars(stmt.returning(User))
>>> print(result.all())
[User(name='spongebob', fullname='Spongebob Squarepants'),
User(name='sandy', fullname='Sandy Cheeks'),
User(name='patrick', fullname='Patrick Star'),
User(name='squidward', fullname='Squidward Tentacles'),
User(name='ehkrabs', fullname='Eugene H. Krabs')]

>>> from sqlalchemy import update
>>> stmt = (
...     update(User)
...     .where(User.name == "squidward")
...     .values(name="spongebob")
...     .returning(User)
... )
>>> result = session.scalars(stmt, execution_options={"synchronize_session": "fetch"})
>>> print(result.all())

>>> users = session.scalars(
...     insert(User).returning(User),
...     [
...         {"name": "spongebob", "fullname": "Spongebob Squarepants"},
...         {"name": "sandy", "fullname": "Sandy Cheeks"},
...         {"name": "patrick", "fullname": "Patrick Star"},
...         {"name": "squidward", "fullname": "Squidward Tentacles"},
...         {"name": "ehkrabs", "fullname": "Eugene H. Krabs"},
...     ],
... )
>>> print(users.all())
[User(name='spongebob', fullname='Spongebob Squarepants'),
 User(name='sandy', fullname='Sandy Cheeks'),
 User(name='patrick', fullname='Patrick Star'),
 User(name='squidward', fullname='Squidward Tentacles'),
 User(name='ehkrabs', fullname='Eugene H. Krabs')]

>>> from sqlalchemy import update
>>> session.execute(
...     update(User),
...     [
...         {"id": 1, "fullname": "Spongebob Squarepants"},
...         {"id": 3, "fullname": "Patrick Star"},
...     ],
... )

>>> from sqlalchemy.dialects.sqlite import insert as sqlite_upsert
>>> stmt = sqlite_upsert(User).values(
...     [
...         {"name": "spongebob", "fullname": "Spongebob Squarepants"},
...         {"name": "sandy", "fullname": "Sandy Cheeks"},
...         {"name": "patrick", "fullname": "Patrick Star"},
...         {"name": "squidward", "fullname": "Squidward Tentacles"},
...         {"name": "ehkrabs", "fullname": "Eugene H. Krabs"},
...     ]
... )
>>> stmt = stmt.on_conflict_do_update(
...     index_elements=[User.name], set_=dict(fullname=stmt.excluded.fullname)
... )
>>> result = session.scalars(stmt.returning(User))
>>> print(result.all())
[User(name='spongebob', fullname='Spongebob Squarepants'),
User(name='sandy', fullname='Sandy Cheeks'),
User(name='patrick', fullname='Patrick Star'),
User(name='squidward', fullname='Squidward Tentacles'),
User(name='ehkrabs', fullname='Eugene H. Krabs')]

>>> from sqlalchemy import update
>>> stmt = (
...     update(User)
...     .where(User.name == "squidward")
...     .values(name="spongebob")
...     .returning(User)
... )
>>> result = session.scalars(stmt, execution_options={"synchronize_session": "fetch"})
>>> print(result.all())

from sqlalchemy.orm import WriteOnlyMapped

class Base(DeclarativeBase):
    pass

class Account(Base):
    __tablename__ = "account"
    id: Mapped[int] = mapped_column(primary_key=True)
    identifier: Mapped[str]
    account_transactions: WriteOnlyMapped["AccountTransaction"] = relationship(
        cascade="all, delete-orphan",
        passive_deletes=True,
        order_by="AccountTransaction.timestamp",
    )

class AccountTransaction(Base):
    __tablename__ = "account_transaction"
    id: Mapped[int] = mapped_column(primary_key=True)
    account_id: Mapped[int] = mapped_column(
        ForeignKey("account.id", ondelete="cascade")
    )
    description: Mapped[str]
    amount: Mapped[Decimal]
    timestamp: Mapped[datetime] = mapped_column(default=func.now())

new_account = Account(
    identifier="account_01",
    account_transactions=[
        AccountTransaction(description="initial deposit", amount=Decimal("500.00")),
        AccountTransaction(description="transfer", amount=Decimal("1000.00")),
        AccountTransaction(description="withdrawal", amount=Decimal("-29.50")),
    ],
)

new_account.account_transactions.add(
    AccountTransaction(description="transfer", amount=Decimal("2000.00"))
)

account_transactions = session.scalars(
    existing_account.account_transactions.select()
    .where(AccountTransaction.amount < 0)
    .limit(10)
).all()

from sqlalchemy.orm import DynamicMapped

class Base(DeclarativeBase):
    pass

class Account(Base):
    __tablename__ = "account"
    id: Mapped[int] = mapped_column(primary_key=True)
    identifier: Mapped[str]
    account_transactions: DynamicMapped["AccountTransaction"] = relationship(
        cascade="all, delete-orphan",
        passive_deletes=True,
        order_by="AccountTransaction.timestamp",
    )

class AccountTransaction(Base):
    __tablename__ = "account_transaction"
    id: Mapped[int] = mapped_column(primary_key=True)
    account_id: Mapped[int] = mapped_column(
        ForeignKey("account.id", ondelete="cascade")
    )
    description: Mapped[str]
    amount: Mapped[Decimal]
    timestamp: Mapped[datetime] = mapped_column(default=func.now())

from sqlalchemy.orm import DynamicMapped

class Base(DeclarativeBase):
    pass

class Account(Base):
    __tablename__ = "account"
    id: Mapped[int] = mapped_column(primary_key=True)
    identifier: Mapped[str]
    account_transactions: DynamicMapped["AccountTransaction"] = relationship(
        cascade="all, delete-orphan",
        passive_deletes=True,
        order_by="AccountTransaction.timestamp",
    )

class AccountTransaction(Base):
    __tablename__ = "account_transaction"
    id: Mapped[int] = mapped_column(primary_key=True)
    account_id: Mapped[int] = mapped_column(
        ForeignKey("account.id", ondelete="cascade")
    )
    description: Mapped[str]
    amount: Mapped[Decimal]
    timestamp: Mapped[datetime] = mapped_column(default=func.now())

meta = MetaData()

my_table = Table(
    "my_table",
    meta,
    Column("id", Integer, primary_key=True),
    Column("num", Integer),
    Column("data", String),
    Index("my_pg_index", "data").ddl_if(dialect="postgresql"),
    CheckConstraint("num > 5").ddl_if(dialect="postgresql"),
)

e1 = create_engine("sqlite://", echo=True)
meta.create_all(e1)  # will not generate CHECK and INDEX

e2 = create_engine("postgresql://scott:tiger@localhost/test", echo=True)
meta.create_all(e2)  # will generate CHECK and INDEX

>>> import datetime

>>> from sqlalchemy import DATETIME
>>> from sqlalchemy import literal
>>> from sqlalchemy.dialects import oracle
>>> from sqlalchemy.dialects import postgresql

>>> date_literal = literal(datetime.datetime.now(), DATETIME)

>>> print(
...     date_literal.compile(
...         dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}
...     )
... )
'2022-12-17 11:02:13.575789'
>>> print(
...     date_literal.compile(
...         dialect=oracle.dialect(), compile_kwargs={"literal_binds": True}
...     )
... )
TO_TIMESTAMP('2022-12-17 11:02:13.575789',  'YYYY-MM-DD HH24:MI:SS.FF') 

with engine.connect() as conn:
    with conn.execution_options(yield_per=100).execute(
        text("select * from table")
    ) as result:
        for row in result:
            print(f"{row}")

async with async_engine.connect() as conn:
    async with conn.execution_options(yield_per=100).execute(
        text("select * from table")
    ) as result:
        for row in result:
            print(f"{row}")

class SomeTest(TestCase):
    def setUp(self):
        # connect to the database
        self.connection = engine.connect()

        # begin a non-ORM transaction
        self.trans = self.connection.begin()

        # bind an individual Session to the connection, selecting
        # "create_savepoint" join_transaction_mode
        self.session = Session(
            bind=self.connection, join_transaction_mode="create_savepoint"
        )

    def tearDown(self):
        self.session.close()

        # rollback non-ORM transaction
        self.trans.rollback()

        # return connection to the Engine
        self.connection.close()

engine = create_engine("...")

# setup outer connection with a transaction and a SAVEPOINT
conn = engine.connect()
trans = conn.begin()
nested = conn.begin_nested()

# bind a Session to that connection and operate upon it, including
# a commit
session = Session(conn)
session.connection()
session.commit()
session.close()

# assert both SAVEPOINT and transaction remain active
assert nested.is_active
nested.rollback()
trans.rollback()

>>> e1 = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")
>>> e2 = create_engine(str(e1.url))

>>> e1 = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")
>>> e2 = create_engine(e1.url)

>>> e1 = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")
>>> url_string = e1.url.render_as_string(hide_password=False)
>>> e2 = create_engine(url_string)

class Foo:
    col1 = mapped_column(Integer)
    col3 = mapped_column(Integer)

class Bar:
    col2 = mapped_column(Integer)
    col4 = mapped_column(Integer)

class Model(Base, Foo, Bar):
    id = mapped_column(Integer, primary_key=True)
    __tablename__ = "model"

CREATE  TABLE  model  (
  col1  INTEGER,
  col3  INTEGER,
  col2  INTEGER,
  col4  INTEGER,
  id  INTEGER  NOT  NULL,
  PRIMARY  KEY  (id)
)

CREATE  TABLE  model  (
  id  INTEGER  NOT  NULL,
  col1  INTEGER,
  col3  INTEGER,
  col2  INTEGER,
  col4  INTEGER,
  PRIMARY  KEY  (id)
)

class Foo:
    id = mapped_column(Integer, primary_key=True)
    col1 = mapped_column(Integer)
    col3 = mapped_column(Integer)

class Model(Foo, Base):
    col2 = mapped_column(Integer)
    col4 = mapped_column(Integer)
    __tablename__ = "model"

CREATE  TABLE  model  (
  col2  INTEGER,
  col4  INTEGER,
  id  INTEGER  NOT  NULL,
  col1  INTEGER,
  col3  INTEGER,
  PRIMARY  KEY  (id)
)

class Foo:
    id = mapped_column(Integer, primary_key=True, sort_order=-10)
    col1 = mapped_column(Integer, sort_order=-1)
    col3 = mapped_column(Integer)

class Model(Foo, Base):
    col2 = mapped_column(Integer)
    col4 = mapped_column(Integer)
    __tablename__ = "model"

CREATE  TABLE  model  (
  id  INTEGER  NOT  NULL,
  col1  INTEGER,
  col2  INTEGER,
  col4  INTEGER,
  col3  INTEGER,
  PRIMARY  KEY  (id)
)

>>> # SQLAlchemy 1.3 (and 2.0)
>>> from sqlalchemy import Sequence
>>> from sqlalchemy.schema import CreateSequence
>>> print(CreateSequence(Sequence("my_seq")))
CREATE  SEQUENCE  my_seq 

>>> # SQLAlchemy 1.4 (only)
>>> from sqlalchemy import Sequence
>>> from sqlalchemy.schema import CreateSequence
>>> print(CreateSequence(Sequence("my_seq")))
CREATE  SEQUENCE  my_seq  START  WITH  1 

>>> # All SQLAlchemy versions
>>> from sqlalchemy import Sequence
>>> from sqlalchemy.schema import CreateSequence
>>> print(CreateSequence(Sequence("my_seq", start=1)))
CREATE  SEQUENCE  my_seq  START  WITH  1 

import typing

from sqlalchemy import String
from sqlalchemy.dialects.mysql import VARCHAR

type_ = String(255).with_variant(VARCHAR(255, charset="utf8mb4"), "mysql", "mariadb")

if typing.TYPE_CHECKING:
    reveal_type(type_)

info: Type of "type_" is "String"

expr = literal(5, Integer) / literal(10, Integer)

%(param_1)s  /  CAST(%(param_2)s  AS  NUMERIC)

expr = literal(5, Integer) // literal(10, Integer)

FLOOR(%(param_1)s  /  %(param_2)s)

expr = func.floor(literal(5, Integer) / literal(10, Integer))

Traceback (most recent call last):
File "/home/classic/dev/sqlalchemy/test3.py", line 30, in worker
    sess.execute(select(A)).all()
File "/home/classic/tmp/sqlalchemy/lib/sqlalchemy/orm/session.py", line 1691, in execute
    conn = self._connection_for_bind(bind)
File "/home/classic/tmp/sqlalchemy/lib/sqlalchemy/orm/session.py", line 1532, in _connection_for_bind
    return self._transaction._connection_for_bind(
File "/home/classic/tmp/sqlalchemy/lib/sqlalchemy/orm/session.py", line 754, in _connection_for_bind
    if self.session.twophase and self._parent is None:
AttributeError: 'NoneType' object has no attribute 'twophase'

File "/home/classic/dev/sqlalchemy/lib/sqlalchemy/orm/session.py", line 1785, in close
   self._close_impl(invalidate=False)
File "/home/classic/dev/sqlalchemy/lib/sqlalchemy/orm/session.py", line 1827, in _close_impl
   transaction.close(invalidate)
File "<string>", line 2, in close
File "/home/classic/dev/sqlalchemy/lib/sqlalchemy/orm/session.py", line 506, in _go
   raise sa_exc.InvalidRequestError(
sqlalchemy.exc.InvalidRequestError: Method 'close()' can't be called here;
method '_connection_for_bind()' is already in progress and this would cause
an unexpected state change to symbol('CLOSED')

from sqlalchemy.dialects import oracle

Table("some_table", metadata, Column("value", oracle.FLOAT(126)))

from sqlalchemy.types import Float
from sqlalchemy.dialects import oracle

Table(
    "some_table",
    metadata,
    Column("value", Float(5).with_variant(oracle.FLOAT(16), "oracle")),
)
```### PostgreSQL 后端的新 RANGE / MULTIRANGE 支持和更改

对于 psycopg2、psycopg3 和 asyncpg 方言，已完全实现了 RANGE / MULTIRANGE 支持。新的支持使用一个新的与后端无关的 SQLAlchemy 特定的 `Range` 对象，不需要使用后端特定的导入或扩展步骤。对于多范围支持，使用 `Range` 对象的列表。

使用先前的 psycopg2 特定类型的代码应修改为使用 `Range`，它提供了兼容的接口。

`Range` 对象还具有与 PostgreSQL 相同的比较支持。到目前为止已经实现了 `Range.contains()` 和 `Range.contained_by()` 方法，它们的工作方式与 PostgreSQL 的 `@>` 和 `<@` 相同。未来的版本可能会增加其他操作符支持。

请参阅 Range and Multirange Types 文档，了解使用这一新功能的背景。

另请参阅

范围和多范围类型

[#7156](https://www.sqlalchemy.org/trac/ticket/7156) [#8706](https://www.sqlalchemy.org/trac/ticket/8706)  ### 在 PostgreSQL 上，`match()` 运算符使用 `plainto_tsquery()` 而不是 `to_tsquery()`

在 PostgreSQL 后端上，`Operators.match()` 函数现在呈现 `col @@ plainto_tsquery(expr)`，而不是 `col @@ to_tsquery()`。`plainto_tsquery()` 接受纯文本，而 `to_tsquery()` 接受专用查询符号，因此与其他后端的兼容性较差。

通过使用 `func` 生成 PostgreSQL 特定函数和 `Operators.bool_op()`（`Operators.op()` 的布尔类型版本）生成任意运算符，可以使用所有 PostgreSQL 搜索函数和操作符，方式与先前版本中提供的方式相同。请参阅 全文搜索 中的示例。

在使用 `Operators.match()` 内的 PG 特定指令的现有 SQLAlchemy 项目应直接使用 `func.to_tsquery()`。要以与 1.4 中相同的形式呈现 SQL，请参阅 使用 match() 进行简单的纯文本匹配 的版本说明。

[#7086](https://www.sqlalchemy.org/trac/ticket/7086)  ### `Session` 的新事务连接模式

“将外部事务加入到会话中”的行为已经修订和改进，允许对 `Session` 如何适应已经建立了事务和可能已经建立了保存点的传入 `Connection` 进行显式控制。新的参数 `Session.join_transaction_mode` 包括一系列选项值，可以以几种方式适应现有的事务，最重要的是允许一个 `Session` 以完全事务性的方式操作，专门使用保存点，同时在所有情况下保持外部启动的事务未提交且处于活动状态，从而允许测试套件回滚在测试中发生的所有更改。

这样做的主要改进是，文档中记录的 将会话加入外部事务（例如测试套件） 的配方，也从 SQLAlchemy 1.3 更改为 1.4，现在简化为不再需要显式使用事件处理程序或任何提及显式保存点；通过使用 `join_transaction_mode="create_savepoint"`，`Session` 将永远不会影响传入事务的状态，而是创建一个保存点（即“嵌套事务”）作为其根事务。

以下是在 将会话加入外部事务（例如测试套件） 中给出的示例的一部分说明；请参阅该部分以获取完整示例：

```py
class SomeTest(TestCase):
    def setUp(self):
        # connect to the database
        self.connection = engine.connect()

        # begin a non-ORM transaction
        self.trans = self.connection.begin()

        # bind an individual Session to the connection, selecting
        # "create_savepoint" join_transaction_mode
        self.session = Session(
            bind=self.connection, join_transaction_mode="create_savepoint"
        )

    def tearDown(self):
        self.session.close()

        # rollback non-ORM transaction
        self.trans.rollback()

        # return connection to the Engine
        self.connection.close()

engine = create_engine("...")

# setup outer connection with a transaction and a SAVEPOINT
conn = engine.connect()
trans = conn.begin()
nested = conn.begin_nested()

# bind a Session to that connection and operate upon it, including
# a commit
session = Session(conn)
session.connection()
session.commit()
session.close()

# assert both SAVEPOINT and transaction remain active
assert nested.is_active
nested.rollback()
trans.rollback()

>>> e1 = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")
>>> e2 = create_engine(str(e1.url))

>>> e1 = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")
>>> e2 = create_engine(e1.url)

>>> e1 = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")
>>> url_string = e1.url.render_as_string(hide_password=False)
>>> e2 = create_engine(url_string)

class Foo:
    col1 = mapped_column(Integer)
    col3 = mapped_column(Integer)

class Bar:
    col2 = mapped_column(Integer)
    col4 = mapped_column(Integer)

class Model(Base, Foo, Bar):
    id = mapped_column(Integer, primary_key=True)
    __tablename__ = "model"

CREATE  TABLE  model  (
  col1  INTEGER,
  col3  INTEGER,
  col2  INTEGER,
  col4  INTEGER,
  id  INTEGER  NOT  NULL,
  PRIMARY  KEY  (id)
)

CREATE  TABLE  model  (
  id  INTEGER  NOT  NULL,
  col1  INTEGER,
  col3  INTEGER,
  col2  INTEGER,
  col4  INTEGER,
  PRIMARY  KEY  (id)
)

class Foo:
    id = mapped_column(Integer, primary_key=True)
    col1 = mapped_column(Integer)
    col3 = mapped_column(Integer)

class Model(Foo, Base):
    col2 = mapped_column(Integer)
    col4 = mapped_column(Integer)
    __tablename__ = "model"

CREATE  TABLE  model  (
  col2  INTEGER,
  col4  INTEGER,
  id  INTEGER  NOT  NULL,
  col1  INTEGER,
  col3  INTEGER,
  PRIMARY  KEY  (id)
)

class Foo:
    id = mapped_column(Integer, primary_key=True, sort_order=-10)
    col1 = mapped_column(Integer, sort_order=-1)
    col3 = mapped_column(Integer)

class Model(Foo, Base):
    col2 = mapped_column(Integer)
    col4 = mapped_column(Integer)
    __tablename__ = "model"

CREATE  TABLE  model  (
  id  INTEGER  NOT  NULL,
  col1  INTEGER,
  col2  INTEGER,
  col4  INTEGER,
  col3  INTEGER,
  PRIMARY  KEY  (id)
)

>>> # SQLAlchemy 1.3 (and 2.0)
>>> from sqlalchemy import Sequence
>>> from sqlalchemy.schema import CreateSequence
>>> print(CreateSequence(Sequence("my_seq")))
CREATE  SEQUENCE  my_seq 

>>> # SQLAlchemy 1.4 (only)
>>> from sqlalchemy import Sequence
>>> from sqlalchemy.schema import CreateSequence
>>> print(CreateSequence(Sequence("my_seq")))
CREATE  SEQUENCE  my_seq  START  WITH  1 

>>> # All SQLAlchemy versions
>>> from sqlalchemy import Sequence
>>> from sqlalchemy.schema import CreateSequence
>>> print(CreateSequence(Sequence("my_seq", start=1)))
CREATE  SEQUENCE  my_seq  START  WITH  1 

import typing

from sqlalchemy import String
from sqlalchemy.dialects.mysql import VARCHAR

type_ = String(255).with_variant(VARCHAR(255, charset="utf8mb4"), "mysql", "mariadb")

if typing.TYPE_CHECKING:
    reveal_type(type_)

info: Type of "type_" is "String"

expr = literal(5, Integer) / literal(10, Integer)

%(param_1)s  /  CAST(%(param_2)s  AS  NUMERIC)

expr = literal(5, Integer) // literal(10, Integer)

FLOOR(%(param_1)s  /  %(param_2)s)

expr = func.floor(literal(5, Integer) / literal(10, Integer))

Traceback (most recent call last):
File "/home/classic/dev/sqlalchemy/test3.py", line 30, in worker
    sess.execute(select(A)).all()
File "/home/classic/tmp/sqlalchemy/lib/sqlalchemy/orm/session.py", line 1691, in execute
    conn = self._connection_for_bind(bind)
File "/home/classic/tmp/sqlalchemy/lib/sqlalchemy/orm/session.py", line 1532, in _connection_for_bind
    return self._transaction._connection_for_bind(
File "/home/classic/tmp/sqlalchemy/lib/sqlalchemy/orm/session.py", line 754, in _connection_for_bind
    if self.session.twophase and self._parent is None:
AttributeError: 'NoneType' object has no attribute 'twophase'

File "/home/classic/dev/sqlalchemy/lib/sqlalchemy/orm/session.py", line 1785, in close
   self._close_impl(invalidate=False)
File "/home/classic/dev/sqlalchemy/lib/sqlalchemy/orm/session.py", line 1827, in _close_impl
   transaction.close(invalidate)
File "<string>", line 2, in close
File "/home/classic/dev/sqlalchemy/lib/sqlalchemy/orm/session.py", line 506, in _go
   raise sa_exc.InvalidRequestError(
sqlalchemy.exc.InvalidRequestError: Method 'close()' can't be called here;
method '_connection_for_bind()' is already in progress and this would cause
an unexpected state change to symbol('CLOSED')

from sqlalchemy.dialects import oracle

Table("some_table", metadata, Column("value", oracle.FLOAT(126)))

from sqlalchemy.types import Float
from sqlalchemy.dialects import oracle

Table(
    "some_table",
    metadata,
    Column("value", Float(5).with_variant(oracle.FLOAT(16), "oracle")),
)

在 SQLAlchemy 版本 2.0 之前的任何版本中，将需要上述形式以提供与后端无关的地板除法。

[#4926](https://www.sqlalchemy.org/trac/ticket/4926)

### Session 主动提出当检测到非法并发或重入访问时

`Session` 现在可以更全面地捕获与多线程或其他并发场景中的非法并发状态更改相关的错误，以及执行意外状态更改的事件钩子。

已知的一个错误是当一个`Session`同时在多个线程中使用时会出现`AttributeError: 'NoneType' object has no attribute 'twophase'`，这完全是神秘的。当一个线程调用`Session.commit()`时，内部调用`SessionTransaction.close()`方法来结束事务上下文，与此同时另一个线程正在运行一个查询，如`Session.execute()`。在`Session.execute()`中，获取当前事务的数据库连接的内部方法首先开始断言会话是“活动的”，但在此断言通过后，同时进行的对`Session.close()`的调用干扰了这种状态，导致上述未定义的条件。

更改将应用到围绕`SessionTransaction`对象的所有更改状态方法的保护措施，以便在上述情况下，`Session.commit()`方法将会失败，因为它将试图将状态更改为在已经进行中的方法期间不允许的状态，而该方法希望获取当前连接以运行数据库查询。

使用在[#7433](https://www.sqlalchemy.org/trac/ticket/7433)中展示的测试脚本，先前的错误案例如下：

```py
Traceback (most recent call last):
File "/home/classic/dev/sqlalchemy/test3.py", line 30, in worker
    sess.execute(select(A)).all()
File "/home/classic/tmp/sqlalchemy/lib/sqlalchemy/orm/session.py", line 1691, in execute
    conn = self._connection_for_bind(bind)
File "/home/classic/tmp/sqlalchemy/lib/sqlalchemy/orm/session.py", line 1532, in _connection_for_bind
    return self._transaction._connection_for_bind(
File "/home/classic/tmp/sqlalchemy/lib/sqlalchemy/orm/session.py", line 754, in _connection_for_bind
    if self.session.twophase and self._parent is None:
AttributeError: 'NoneType' object has no attribute 'twophase'

File "/home/classic/dev/sqlalchemy/lib/sqlalchemy/orm/session.py", line 1785, in close
   self._close_impl(invalidate=False)
File "/home/classic/dev/sqlalchemy/lib/sqlalchemy/orm/session.py", line 1827, in _close_impl
   transaction.close(invalidate)
File "<string>", line 2, in close
File "/home/classic/dev/sqlalchemy/lib/sqlalchemy/orm/session.py", line 506, in _go
   raise sa_exc.InvalidRequestError(
sqlalchemy.exc.InvalidRequestError: Method 'close()' can't be called here;
method '_connection_for_bind()' is already in progress and this would cause
an unexpected state change to symbol('CLOSED')

from sqlalchemy.dialects import oracle

Table("some_table", metadata, Column("value", oracle.FLOAT(126)))

from sqlalchemy.types import Float
from sqlalchemy.dialects import oracle

Table(
    "some_table",
    metadata,
    Column("value", Float(5).with_variant(oracle.FLOAT(16), "oracle")),
)