Python 过滤 Sqlalchemy 关系条件

发布于09月11日

我想根据他们的code和mobile number来查询一个user.

值得注意的是，我在连接到数据库时使用的是异步会话.

这是我的机型和疑问:

class model_user(Base):
    __tablename__ = "users"
    id = Column(UUID)
    code = Column(Integer)
    user_name = Column(String(11), unique=True)
    first_name = Column(String(255), index=True)
    last_name = Column(String(255), index=True)
    contact = relationship("model_contact", back_populates="user")

class model_contact(Base):
    __tablename__ = "contacts"
    id = Column(Integer)
    mobile = Column(String(320))
    user_id = Column(
        UUID(as_uuid=True), ForeignKey("users.id"), nullable=False, index=True
    )
    user = relationship("model_user", back_populates="contact")


# Here is my query:
query = await db.execute(
    select(user_model)\
    .filter_by(code==5)\
    .options(
        joinedload(model_user.contact)
    )\
    .filter(model_contact.mobile == mobile)

推荐答案

这实际上与您是否使用SQLAlChemy的async扩展没有任何关系.查询的构造方式相同.只是会话设置和交互明显不同.

附注:

您应该使用PascalCase来命名您的类，并且在名称中包含术语"模型"通常不是很好的风格，例如User和Contact.
由于您的用户模型和联系人模型之间存在一对多关系(即一个用户可以有多组联系人信息)，因此您应该使用复数来命名用户模型上的关系属性，即contacts.

我能想到的做你想做的事情的最简单的方法是使用Select.where方法.然后，您可以使用关系的any方法构造SQL EXISTS子查询.声明将如下所示:

statement = select(User).where(
    User.code == 123,
    User.contacts.any(Contact.mobile == "555")
)

下面是一个使用aiosqlite的完整工作示例，以证明它与async个工具一起工作:

from asyncio import run

from sqlalchemy import Column, ForeignKey, Integer, String, select
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, sessionmaker


Base = declarative_base()


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    code = Column(Integer)
    contacts = relationship("Contact", back_populates="user")


class Contact(Base):
    __tablename__ = "contacts"
    id = Column(Integer, primary_key=True)
    mobile = Column(String(320))
    user_id = Column(
        Integer(), ForeignKey("users.id"), nullable=False, index=True
    )
    user = relationship(User, back_populates="contacts")


async def main():
    engine = create_async_engine("sqlite+aiosqlite://", echo=True)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    async_session = sessionmaker(
        engine, expire_on_commit=False, class_=AsyncSession
    )
    async with async_session() as session:
        statement = select(User).where(
            User.code == 123,
            User.contacts.any(Contact.mobile == "555")
        )
        await session.execute(statement)


if __name__ == "__main__":
    run(main())

运行此脚本将为select查询生成以下SQL输出:

SELECT users.id, users.code 
FROM users 
WHERE users.code = ? AND (EXISTS (SELECT 1 
FROM contacts 
WHERE users.id = contacts.user_id AND contacts.mobile = ?))
...
(123, '555')

这种方法应该会给您带来您想要的结果.