SqlAlchemy 2.0 中文文档（七十三）（1）

j = join(B, D, D.b_id == B.id).join(C, C.id == D.c_id)
B_viacd = mapper(
    B,
    j,
    non_primary=True,
    primary_key=[j.c.b_id],
    properties={
        "id": j.c.b_id,  # so that 'id' looks the same as before
        "c_id": j.c.c_id,  # needed for disambiguation
        "d_c_id": j.c.d_c_id,  # needed for disambiguation
        "b_id": [j.c.b_id, j.c.d_b_id],
        "d_id": j.c.d_id,
    },
)
A.b = relationship(B_viacd, primaryjoin=A.b_id == B_viacd.c.b_id)

j = join(B, D, D.b_id == B.id).join(C, C.id == D.c_id)
B_viacd = aliased(B, j, flat=True)
A.b = relationship(B_viacd, primaryjoin=A.b_id == j.c.b_id)

class A(Base):
    __tablename__ = "a"
    id = Column(Integer, primary_key=True)
    bs = relationship("B", lazy="selectin")
class B(Base):
    __tablename__ = "b"
    id = Column(Integer, primary_key=True)
    a_id = Column(ForeignKey("a.id"))

SELECT  a.id  AS  a_id  FROM  a
SELECT  a_1.id  AS  a_1_id,  b.id  AS  b_id,  b.a_id  AS  b_a_id
FROM  a  AS  a_1  JOIN  b  ON  a_1.id  =  b.a_id
WHERE  a_1.id  IN  (?,  ?,  ?,  ?,  ?,  ?,  ?,  ?,  ?,  ?)  ORDER  BY  a_1.id
(1,  2,  3,  4,  5,  6,  7,  8,  9,  10)

SELECT  a.id  AS  a_id  FROM  a
SELECT  b.a_id  AS  b_a_id,  b.id  AS  b_id  FROM  b
WHERE  b.a_id  IN  (?,  ?,  ?,  ?,  ?,  ?,  ?,  ?,  ?,  ?)  ORDER  BY  b.a_id
(1,  2,  3,  4,  5,  6,  7,  8,  9,  10)

u1 = session.query(User).get(5)
query = session.query(Address).filter(Address.user == u1)

u1 = session.query(User).get(5)
query = session.query(Address).filter(Address.user == u1)
session.expire(u1)
session.expunge(u1)
query.all()  # <-- would raise DetachedInstanceError

a1 = session.query(Address).filter_by(id=5).one()
session.expunge(a1)
a1.user = some_user

class Address(Base):
    # ...
    user = relationship("User", ..., lazy="raise")

some_object = session.query(SomeObject).get(5)
del some_object.some_attribute  # from a SQL perspective, works like "= None"

from sqlalchemy import inspect
u1 = User(id=7, name="ed")
inspect(u1).info["user_info"] = "7|ed"

class A(Base):
    __tablename__ = "test_a"
    id = Column(Integer, primary_key=True)
    ab = relationship("AB", backref="a", uselist=False)
    b = association_proxy(
        "ab", "b", creator=lambda b: AB(b=b), cascade_scalar_deletes=True
    )
class B(Base):
    __tablename__ = "test_b"
    id = Column(Integer, primary_key=True)
    ab = relationship("AB", backref="b", cascade="all, delete-orphan")
class AB(Base):
    __tablename__ = "test_ab"
    a_id = Column(Integer, ForeignKey(A.id), primary_key=True)
    b_id = Column(Integer, ForeignKey(B.id), primary_key=True)

a.b = B()

a.b = None
assert a.ab is None

del a.b
assert a.ab is None

class User(Base):
    # ...
    keywords = association_proxy("kws", "keyword")
proxy_state = inspect(User).all_orm_descriptors["keywords"].for_class(User)

>>> proxy_state.target_class
Keyword

class User(Base):
    # ...
    elements = relationship("Element")
    # column-based association proxy
    values = association_proxy("elements", "value")
class Element(Base):
    # ...
    value = Column(String)

>>> print(s.query(User).filter(User.values.like("%foo%")))
SELECT  "user".id  AS  user_id
FROM  "user"
WHERE  EXISTS  (SELECT  1
FROM  element
WHERE  "user".id  =  element.user_id  AND  element.value  LIKE  :value_1) 

>>> print(s.query(User).filter(User.values == "foo"))
SELECT  "user".id  AS  user_id
FROM  "user"
WHERE  EXISTS  (SELECT  1
FROM  element
WHERE  "user".id  =  element.user_id  AND  element.value  =  :value_1) 

>>> print(s.query(User).filter(User.values == None))
SELECT  "user".id  AS  user_id
FROM  "user"
WHERE  (EXISTS  (SELECT  1
FROM  element
WHERE  "user".id  =  element.user_id  AND  element.value  IS  NULL))  OR  NOT  (EXISTS  (SELECT  1
FROM  element
WHERE  "user".id  =  element.user_id)) 

>>> print(s.query(User).filter(User.values.contains("foo")))
SELECT  "user".id  AS  user_id
FROM  "user"
WHERE  EXISTS  (SELECT  1
FROM  element
WHERE  "user".id  =  element.user_id  AND  (element.value  LIKE  '%'  ||  :value_1  ||  '%')) 

class User(Base):
    __tablename__ = "user"
    id = Column(Integer, primary_key=True)
    user_elements = relationship("UserElement")
    # object-based association proxy
    elements = association_proxy("user_elements", "element")
class UserElement(Base):
    __tablename__ = "user_element"
    id = Column(Integer, primary_key=True)
    user_id = Column(ForeignKey("user.id"))
    element_id = Column(ForeignKey("element.id"))
    element = relationship("Element")
class Element(Base):
    __tablename__ = "element"
    id = Column(Integer, primary_key=True)
    value = Column(String)

>>> print(s.query(User).filter(User.elements.contains(Element(id=1))))
SELECT "user".id AS user_id
FROM "user"
WHERE EXISTS (SELECT 1
FROM user_element
WHERE "user".id = user_element.user_id AND :param_1 = user_element.element_id)

class A(Base):
    __tablename__ = "a"
    id = Column(Integer, primary_key=True)
    bs = relationship("B")
    b_data = association_proxy("bs", "data")
class B(Base):
    __tablename__ = "b"
    id = Column(Integer, primary_key=True)
    a_id = Column(ForeignKey("a.id"))
    data = Column(String)
a1 = A(bs=[B(data="b1"), B(data="b2")])
b_data = a1.b_data

del a1

collection = session.query(A).filter_by(id=1).first().b_data

assert b_data == ["b1", "b2"]

class A(Base):
    __tablename__ = "test_a"
    id = Column(Integer, primary_key=True)
    b_rel = relationship(
        "B",
        collection_class=set,
        cascade="all, delete-orphan",
    )
    b = association_proxy("b_rel", "value", creator=lambda x: B(value=x))
class B(Base):
    __tablename__ = "test_b"
    __table_args__ = (UniqueConstraint("a_id", "value"),)
    id = Column(Integer, primary_key=True)
    a_id = Column(Integer, ForeignKey("test_a.id"), nullable=False)
    value = Column(String)
# ...
s = Session(e)
a = A(b={"x", "y", "z"})
s.add(a)
s.commit()
# re-assign where one B should be deleted, one B added, two
# B's maintained
a.b = {"x", "z", "q"}
# only 'q' was added, so only one new B object.  previously
# all three would have been re-created leading to flush conflicts
# against the deleted ones.
assert len(s.new) == 1

class A(Base):
    __tablename__ = "a"
    id = Column(Integer, primary_key=True)
    bs = relationship("B", backref="a")
class B(Base):
    __tablename__ = "b"
    id = Column(Integer, primary_key=True)
    a_id = Column(ForeignKey("a.id"))

a1 = A(bs=[B(), B()])
a1.bs[0], a1.bs[1] = a1.bs[1], a1.bs[0]

>>> a1 = A()
>>> b1 = B()
>>> a1.bs.append(b1)
>>> a1.bs.append(b1)  # append the same b1 object twice
>>> del a1.bs[1]
>>> a1.bs  # collection is unaffected so far...
[<__main__.B object at 0x7f047af5fb70>]
>>> b1.a  # however b1.a is None
>>>
>>> session.add(a1)
>>> session.commit()  # so upon flush + expire....
>>> a1.bs  # the value is gone
[]

u_alias = aliased(User)
session.query(User, u_alias).join(Address)

SELECT  ...
FROM  users  AS  users_1,  users  JOIN  addresses  ON  users.id  =  addresses.user_id

sqlalchemy.exc.InvalidRequestError: Can't determine which FROM clause to
join from, there are multiple FROMS which can join to this entity.
Try adding an explicit ON clause to help resolve the ambiguity.

# join to User
session.query(User, u_alias).join(Address, Address.user_id == User.id)
# join to u_alias
session.query(User, u_alias).join(Address, Address.user_id == u_alias.id)

# join to User
session.query(User, u_alias).join(Address, User.addresses)
# join to u_alias
session.query(User, u_alias).join(Address, u_alias.addresses)

session.query(func.current_timestamp(), User).join(Address)

sqlalchemy.exc.InvalidRequestError: Don't know how to join from
CURRENT_TIMESTAMP; please use select_from() to establish the
left entity/selectable of this join

SELECT  CURRENT_TIMESTAMP  AS  current_timestamp_1,  users.id  AS  users_id,
users.name  AS  users_name,  users.fullname  AS  users_fullname,
users.password  AS  users_password
FROM  users  JOIN  addresses  ON  users.id  =  addresses.user_id

session.query(A).options(joinedload(A.b)).limit(5)

SELECT  subq.a_id,  subq.a_data,  b_alias.id,  b_alias.data  FROM  (
  SELECT  a.id  AS  a_id,  a.data  AS  a_data  FROM  a  LIMIT  5
)  AS  subq  LEFT  OUTER  JOIN  b  ON  subq.a_id=b.a_id

SELECT  subq.a_id,  subq.a_data,  b_alias.id,  b_alias.data  FROM  (
  SELECT  a.id  AS  a_id,  a.data  AS  a_data  FROM  a  LIMIT  5
)  AS  subq  LEFT  OUTER  JOIN  b  ON  subq.a_id=b.a_id  FOR  UPDATE

SELECT  subq.a_id,  subq.a_data,  b_alias.id,  b_alias.data  FROM  (
  SELECT  a.id  AS  a_id,  a.data  AS  a_data  FROM  a  LIMIT  5  FOR  UPDATE
)  AS  subq  LEFT  OUTER  JOIN  b  ON  subq.a_id=b.a_id  FOR  UPDATE

session.query(A).options(joinedload(A.b)).with_for_update(of=A).limit(5)

SELECT  subq.a_id,  subq.a_data,  b_alias.id,  b_alias.data  FROM  (
  SELECT  a.id  AS  a_id,  a.data  AS  a_data  FROM  a  LIMIT  5  FOR  UPDATE  OF  a
)  AS  subq  LEFT  OUTER  JOIN  b  ON  subq.a_id=b.a_id

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    addresses = relationship("Address", passive_deletes="all")
class Address(Base):
    __tablename__ = "addresses"
    id = Column(Integer, primary_key=True)
    email = Column(String)
    user_id = Column(Integer, ForeignKey("users.id"))
    user = relationship("User")
u1 = session.query(User).first()
address = u1.addresses[0]
u1.addresses.remove(address)
session.commit()
# would fail and be set to None
assert address.user_id == u1.id

metadata_obj = MetaData(
    naming_convention={"uq": "uq_%(table_name)s_%(column_0_N_name)s"}
)
table = Table(
    "info",
    metadata_obj,
    Column("a", Integer),
    Column("b", Integer),
    Column("c", Integer),
    UniqueConstraint("a", "b", "c"),
)

CREATE  TABLE  info  (
  a  INTEGER,
  b  INTEGER,
  c  INTEGER,
  CONSTRAINT  uq_info_a_b_c  UNIQUE  (a,  b,  c)
)

long_names = Table(
    "long_names",
    metadata_obj,
    Column("information_channel_code", Integer, key="a"),
    Column("billing_convention_name", Integer, key="b"),
    Column("product_identifier", Integer, key="c"),
    UniqueConstraint("a", "b", "c"),
)

CREATE  TABLE  long_names  (
  information_channel_code  INTEGER,
  billing_convention_name  INTEGER,
  product_identifier  INTEGER,
  CONSTRAINT  uq_long_names_information_channel_code_billing_conventi_a79e
  UNIQUE  (information_channel_code,  billing_convention_name,  product_identifier)
)

from sqlalchemy import Column
from sqlalchemy import Integer
from sqlalchemy import MetaData
from sqlalchemy import Table
from sqlalchemy import UniqueConstraint
from sqlalchemy.dialects import postgresql
from sqlalchemy.schema import AddConstraint
m = MetaData()
t = Table("t", m, Column("x", Integer))
uq = UniqueConstraint(
    t.c.x,
    name="this_is_too_long_of_a_name_for_any_database_backend_even_postgresql",
)
print(AddConstraint(uq).compile(dialect=postgresql.dialect()))

sqlalchemy.exc.IdentifierError: Identifier
'this_is_too_long_of_a_name_for_any_database_backend_even_postgresql'
exceeds maximum length of 63 characters

uq = UniqueConstraint(
    t.c.x,
    name=conv("this_is_too_long_of_a_name_for_any_database_backend_even_postgresql"),
)

ALTER  TABLE  t  ADD  CONSTRAINT  this_is_too_long_of_a_name_for_any_database_backend_eve_ac05  UNIQUE  (x)

class Venue(Base):
    __tablename__ = "venue"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    descendants = relationship(
        "Venue",
        primaryjoin=func.instr(remote(foreign(name)), name + "/").as_comparison(1, 2)
        == 1,
        viewonly=True,
        order_by=name,
    )

SELECT  venue.id  AS  venue_id,  venue.name  AS  venue_name
FROM  venue
WHERE  instr(venue.name,  (?  ||  ?))  =  ?  ORDER  BY  venue.name
('parent1',  '/',  1)

v1 = (
    s.query(Venue)
    .filter_by(name="parent1")
    .options(joinedload(Venue.descendants))
    .one()
)

SELECT  venue.id  AS  venue_id,  venue.name  AS  venue_name,
  venue_1.id  AS  venue_1_id,  venue_1.name  AS  venue_1_name
FROM  venue  LEFT  OUTER  JOIN  venue  AS  venue_1
  ON  instr(venue_1.name,  (venue.name  ||  ?))  =  ?
WHERE  venue.name  =  ?  ORDER  BY  venue_1.name
('/',  1,  'parent1')

>>> from sqlalchemy import create_engine
>>> from sqlalchemy import select, literal_column, bindparam
>>> e = create_engine("postgresql://scott:tiger@localhost/test", echo=True)
>>> with e.connect() as conn:
...     conn.execute(
...         select([literal_column("1")]).where(
...             literal_column("1").in_(bindparam("q", expanding=True))
...         ),
...         q=[],
...     )
{exexsql}SELECT 1 WHERE 1 IN (SELECT CAST(NULL AS INTEGER) WHERE 1!=1)

>>> from sqlalchemy import create_engine
>>> from sqlalchemy import select, literal_column, tuple_, bindparam
>>> e = create_engine("postgresql://scott:tiger@localhost/test", echo=True)
>>> with e.connect() as conn:
...     conn.execute(
...         select([literal_column("1")]).where(
...             tuple_(50, "somestring").in_(bindparam("q", expanding=True))
...         ),
...         q=[],
...     )
{exexsql}SELECT 1 WHERE (%(param_1)s, %(param_2)s)
IN (SELECT CAST(NULL AS INTEGER), CAST(NULL AS VARCHAR) WHERE 1!=1)

from sqlalchemy import TypeDecorator, LargeBinary, func
class CompressedLargeBinary(TypeDecorator):
    impl = LargeBinary
    def bind_expression(self, bindvalue):
        return func.compress(bindvalue, type_=self)
    def column_expression(self, col):
        return func.uncompress(col, type_=self)
MyLargeBinary = LargeBinary().with_variant(CompressedLargeBinary(), "sqlite")

from sqlalchemy import select, column
from sqlalchemy.dialects import sqlite
print(select([column("x", CompressedLargeBinary)]).compile(dialect=sqlite.dialect()))

SELECT  uncompress(x)  AS  x