基于code和手机号查询user的异步SQLAlchemy查询问题

模型定义的问题与修正 #

你的模型存在几处需要完善的细节：

• model_user和model_contact的id字段未设置主键，数据库表必须有主键才能保证数据唯一性与查询效率
• model_user的UUID类型字段建议添加默认值，避免手动生成ID
• model_contact的mobile字段用于查询，建议添加索引提升查询速度

修正后的模型代码：

import uuid
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship, Base

class model_user(Base):
    __tablename__ = "users"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    code = Column(Integer, index=True)  # 给查询字段加索引
    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, primary_key=True, autoincrement=True)
    mobile = Column(String(320), index=True)  # 手机号查询加索引
    user_id = Column(
        UUID(as_uuid=True), ForeignKey("users.id"), nullable=False, index=True
    )
    user = relationship("model_user", back_populates="contact")

查询语句的问题与优化 #

原查询存在两个关键错误：

1. filter_by(code==5)写法错误：filter_by接收关键字参数，应写成filter_by(code=5)，不能用==
2. 直接过滤model_contact.mobile会报错：joinedload仅用于预加载关联数据，不会将关联表加入查询的FROM子句，必须显式JOIN关联表才能使用其字段过滤

优化后的异步查询提供两种可选方案：

方案一：显式JOIN关联表 #

适合需要同时获取用户和关联联系方式的场景：

# 假设user_model是model_user的引用，mobile是传入的手机号变量
result = await db.execute(
    select(model_user)
    .join(model_contact, model_user.id == model_contact.user_id)
    .filter(model_user.code == 5)
    .filter(model_contact.mobile == mobile)
    .options(joinedload(model_user.contact))  # 预加载避免N+1查询
)
user = result.scalar_one_or_none()  # 获取单个用户，无结果返回None

方案二：使用EXISTS子查询 #

适合仅需获取用户数据、不需要联系方式详情的场景，性能更优：

from sqlalchemy import exists

subquery = exists().where(
    (model_contact.user_id == model_user.id) & (model_contact.mobile == mobile)
)

result = await db.execute(
    select(model_user)
    .filter(model_user.code == 5)
    .filter(subquery)
)
user = result.scalar_one_or_none()

额外提示 #

• 异步会话中，使用scalar_one_or_none()获取结果比直接处理Result对象更简洁
• 如果code是高频查询字段，给其添加索引能显著提升查询速度（已在修正后的模型中添加）

内容的提问来源于stack exchange，提问作者Masy