我有一个模式,其中包括自联接和多对多关系,如下所示:
from typing import List, Optional
from sqlalchemy import create_engine
from sqlalchemy.orm import (
aliased,
DeclarativeBase,
Session,
Mapped,
mapped_column,
relationship,
)
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.schema import ForeignKey
from sqlalchemy.types import String
class Base(DeclarativeBase):
pass
class Post(Base):
__tablename__ = 'post'
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str] = mapped_column(String(200), nullable=False)
parent_id: Mapped[int] = mapped_column(ForeignKey('post.id'), nullable=True)
parent: Mapped["Post"] = relationship('Post', foreign_keys=parent_id, remote_side=id)
tags: Mapped[List['Tag']] = relationship('Tag', secondary='tag2post')
class Tag(Base):
__tablename__ = 'tag'
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100), nullable=False)
class Tag2Post(Base):
__tablename__ = 'tag2post'
id: Mapped[int] = mapped_column(primary_key=True)
tag_id: Mapped[int] = mapped_column('tag_id', ForeignKey('tag.id'))
tag: Mapped[Tag] = relationship(Tag, overlaps='tags')
post_id: Mapped[int] = mapped_column('post_id', ForeignKey('post.id'))
post: Mapped[Post] = relationship(Post, overlaps='tags')
engine = create_engine("sqlite+pysqlite:///:memory:", echo=True)
Base.metadata.create_all(engine)
with Session(engine) as session:
session.add(tag_a := Tag(name='a'))
session.add(tag_b := Tag(name='b'))
session.add(parent := Post(title='parent', tags=[tag_a]))
session.add(child := Post(title='child', parent=parent, tags=[tag_b]))
现在我想编写一个以下形式的查询“给我所有带有标签
a
并且其父级有一个标记
b
”。
我了解为每个步骤添加别名的“手动”方式:
parent_alias = aliased(Post)
parent_tag = aliased(Tag)
parent_tag2post = aliased(Tag2Post)
q = session.query(
Post
).join(
parent_alias, Post.parent_id == parent_alias.id
).join(
parent_tag2post,
parent_alias.id == parent_tag2post.post_id
).join(
parent_tag,
parent_tag2post.tag_id == parent_tag.id,
).join(
Post.tags,
).filter(
parent_tag.name == 'a',
Tag.name == 'b',
)
print(q.one().title) # prints 'child'
但我正在处理更通用的代码,基本上是一种为我提供条件的查询语言:| ||将查询语言转换为联接的代码现在需要内省关系(如
-
Post.tags.name == 'b'
-
Post.parent.tags.name == 'a'
、
Post.tags
、
Post.parents
),并在每个关系中用别名替换表并用别名重建联接条件...这听起来相当复杂且容易出错。
Post.parents.tags
有更简单的方法吗?也许,就像告诉 sqlalchemy 使用关系,但使用别名?或者是否有第三方软件包已经可以做到这一点?
Is there an easier way? Like, maybe, telling sqlalchemy to use a relationship, but with an alias? Or is there maybe a third-party package that already does this?
的确,手动添加别名来处理 SQLAlchemy 中复杂的关系可能会很繁琐且容易出错。幸运的是,SQLAlchemy 提供了一些机制可以简化这个过程。
1. 使用
any
和
has
进行关联表查询
对于的查询“获取所有带有标签 'a' 且其父级带有标签 'b' 的帖子”,可以使用 SQLAlchemy 的
any
和
has
操作符来简化代码,而无需手动创建别名:
from sqlalchemy import and_
q = session.query(Post).\
filter(Post.tags.any(Tag.name == 'a')).\
filter(Post.parent.has(Post.tags.any(Tag.name == 'b')))
print(q.one().title) # prints 'child'
-
Post.tags.any(Tag.name == 'a')
: 这部分检查 Post 是否与名称为 'a' 的 Tag 相关联。 -
Post.parent.has(Post.tags.any(Tag.name == 'b'))
: 这部分检查 Post 的父级(也是一个 Post 对象)是否与名称为 'b' 的 Tag 相关联。
这种方式更加简洁易懂,因为它直接表达了查询的语义,而无需手动处理别名。
2. 使用
with_entities
加载关联数据
如果需要在查询结果中同时获取帖子及其关联的标签和父级信息,可以使用
with_entities
方法,并利用 SQLAlchemy 的自动别名功能:
q = session.query(Post).\
options(
joinedload(Post.parent).
joinedload(Post.parent.tags)
).\
filter(Post.tags.any(Tag.name == 'a')).\
filter(Post.parent.has(Post.tags.any(Tag.name == 'b'))).\
with_entities(Post, Post.parent, Tag)
results = q.all()
for post, parent, tag in results:
print(f"Post: {post.title}, Parent: {parent.title if parent else None}, Tag: {tag.name}")
-
joinedload(Post.parent).joinedload(Post.parent.tags)
: 使用joinedload
可以一次性加载关联的父级及其标签,避免 N+1 查询问题。 -
with_entities(Post, Post.parent, Tag)
: 指定要查询的实体,SQLAlchemy 会自动为关联的实体创建别名。
3. 使用第三方库:SQLAlchemy-Utils
如果需要更强大的查询构建功能,可以考虑使用第三方库,例如 SQLAlchemy-Utils。该库提供了一些工具来简化复杂查询的构建,包括自动别名管理。
总而言之,对于的情况,使用
any
和
has
操作符结合
with_entities
以及
joinedload
可以更简洁地表达查询,而无需手动处理别名。如果需要更通用的查询构建机制,可以考虑使用 SQLAlchemy-Utils 等第三方库。