SqlAlchemy 2.0 中文文档（五十五）（2）

class A(Base):
    __tablename__ = "a"
    id = Column(Integer, primary_key=True)
    bs = relationship("B", back_populates="a")
class B(Base):
    __tablename__ = "b"
    id = Column(Integer, primary_key=True)
    a_id = Column(ForeignKey("a.id"))
    # this will emit the error message when the mapper
    # configuration step occurs
    a = relationship("A", back_populates="bs", cascade="all, delete-orphan")
configure_mappers()

class A(Base):
    __tablename__ = "a"
    id = Column(Integer, primary_key=True)
    bs = relationship("B", back_populates="a", cascade="all, delete-orphan")
class B(Base):
    __tablename__ = "b"
    id = Column(Integer, primary_key=True)
    a_id = Column(ForeignKey("a.id"))
    a = relationship("A", back_populates="bs")

class A(Base):
    __tablename__ = "a"
    id = Column(Integer, primary_key=True)
    bs = relationship("B", back_populates="a")
class B(Base):
    __tablename__ = "b"
    id = Column(Integer, primary_key=True)
    a_id = Column(ForeignKey("a.id"))
    a = relationship(
        "A",
        back_populates="bs",
        single_parent=True,
        cascade="all, delete-orphan",
    )

>>> b1 = B()
>>> b2 = B()
>>> a1 = A()
>>> b1.a = a1
>>> b2.a = a1
sqlalchemy.exc.InvalidRequestError: Instance <A at 0x7eff44359350> is
already associated with an instance of <class '__main__.B'> via its
B.a attribute, and is only allowed a single parent.

>>> a1.bs = [b1, b2]
>>> session.add_all([a1, b1, b2])
>>> session.commit()
INSERT  INTO  a  DEFAULT  VALUES
()
INSERT  INTO  b  (a_id)  VALUES  (?)
(1,)
INSERT  INTO  b  (a_id)  VALUES  (?)
(1,) 

>>> session.delete(b1)
>>> session.commit()
UPDATE  b  SET  a_id=?  WHERE  b.id  =  ?
(None,  2)
DELETE  FROM  b  WHERE  b.id  =  ?
(1,)
DELETE  FROM  a  WHERE  a.id  =  ?
(1,)
COMMIT 

class A(Base):
    __tablename__ = "a"
    id = Column(Integer, primary_key=True)
class B(Base):
    __tablename__ = "b"
    id = Column(Integer, primary_key=True)
    a_id = Column(ForeignKey("a.id"))
    a = relationship(
        "A",
        single_parent=True,
        cascade="all, delete-orphan",
    )

>>> b1 = B()
>>> b2 = B()
>>> a1 = A()
>>> b1.a = a1
>>> b2.a = a1
sqlalchemy.exc.InvalidRequestError: Instance <A at 0x7eff44359350> is
already associated with an instance of <class '__main__.B'> via its
B.a attribute, and is only allowed a single parent.

class Parent(Base):
    __tablename__ = "parent"
    id = Column(Integer, primary_key=True)
    children = relationship("Child")
class Child(Base):
    __tablename__ = "child"
    id = Column(Integer, primary_key=True)
    parent_id = Column(ForeignKey("parent.id"))
    parent = relationship("Parent")

SAWarning: relationship 'Child.parent' will copy column parent.id to column child.parent_id,
which conflicts with relationship(s): 'Parent.children' (copies parent.id to child.parent_id).

class Parent(Base):
    __tablename__ = "parent"
    id = Column(Integer, primary_key=True)
    children = relationship("Child", back_populates="parent")
class Child(Base):
    __tablename__ = "child"
    id = Column(Integer, primary_key=True)
    parent_id = Column(ForeignKey("parent.id"))
    parent = relationship("Parent", back_populates="children")

class Parent(Base):
    __tablename__ = "parent"
    id = Column(Integer, primary_key=True)
    c1 = relationship(
        "Child",
        primaryjoin="and_(Parent.id == Child.parent_id, Child.flag == 0)",
        backref="parent",
        overlaps="c2, parent",
    )
    c2 = relationship(
        "Child",
        primaryjoin="and_(Parent.id == Child.parent_id, Child.flag == 1)",
        overlaps="c1, parent",
    )
class Child(Base):
    __tablename__ = "child"
    id = Column(Integer, primary_key=True)
    parent_id = Column(ForeignKey("parent.id"))
    flag = Column(Integer)

# context manager creates new Session
with Session(engine) as session_obj:
    result = sess.execute(select(User).where(User.id == 7))
# context manager is closed, so session_obj above is closed, identity
# map is replaced
# iterating the result object can't associate the object with the
# Session, raises this error.
user = result.first()

# context manager creates new Session
with Session(engine) as session_obj:
    # result internally pre-fetches all objects
    result = sess.execute(
        select(User).where(User.id == 7), execution_options={"prebuffer_rows": True}
    )
# context manager is closed, so session_obj above is closed, identity
# map is replaced
# pre-buffered objects are returned
user = result.first()
# however they are detached from the session, which has been closed
assert inspect(user).detached
assert inspect(user).session is None

from __future__ import annotations
import inspect
from typing import Optional
from uuid import uuid4
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import MappedAsDataclass
class Mixin:
    create_user: Mapped[int] = mapped_column()
    update_user: Mapped[Optional[int]] = mapped_column(default=None, init=False)
class Base(DeclarativeBase, MappedAsDataclass):
    pass
class User(Base, Mixin):
    __tablename__ = "sys_user"
    uid: Mapped[str] = mapped_column(
        String(50), init=False, default_factory=uuid4, primary_key=True
    )
    username: Mapped[str] = mapped_column()
    email: Mapped[str] = mapped_column()

SADeprecationWarning: When transforming <class '__main__.User'> to a
dataclass, attribute(s) "create_user", "update_user" originates from
superclass <class
'__main__.Mixin'>, which is not a dataclass. This usage is deprecated and
will raise an error in SQLAlchemy 2.1\. When declaring SQLAlchemy
Declarative Dataclasses, ensure that all mixin classes and other
superclasses which include attributes are also a subclass of
MappedAsDataclass.

class Mixin(MappedAsDataclass):
    create_user: Mapped[int] = mapped_column()
    update_user: Mapped[Optional[int]] = mapped_column(default=None, init=False)

from dataclasses import dataclass
from dataclasses import field
import inspect
from typing import Optional
from uuid import uuid4
class Mixin:
    create_user: int
    update_user: Optional[int] = field(default=None)
@dataclass
class User(Mixin):
    uid: str = field(init=False, default_factory=lambda: str(uuid4()))
    username: str
    password: str
    email: str

>>> session.execute(
...     update(User).where(User.name == bindparam("u_name")),
...     [
...         {"u_name": "spongebob", "fullname": "Spongebob Squarepants"},
...         {"u_name": "patrick", "fullname": "Patrick Star"},
...     ],
... )

>>> session.execute(
...     update(User),
...     [
...         {"id": 1, "fullname": "Spongebob Squarepants"},
...         {"id": 3, "fullname": "Patrick Star"},
...         {"id": 5, "fullname": "Eugene H. Krabs"},
...     ],
... )

>>> session.connection().execute(
...     update(User).where(User.name == bindparam("u_name")),
...     [
...         {"u_name": "spongebob", "fullname": "Spongebob Squarepants"},
...         {"u_name": "patrick", "fullname": "Patrick Star"},
...     ],
... )