将PostgreSQL查询转换为SQLAlchemy实现求助

首先假设你已经定义了匹配数据库表结构的SQLAlchemy模型：

from sqlalchemy import Column, Integer, String, ForeignKey, func, exists, literal
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import select

Base = declarative_base()

class Skill(Base):
    __tablename__ = 'skill'
    __table_args__ = {'schema': 'backend'}
    id_skill = Column(Integer, primary_key=True)
    skill_name = Column(String, nullable=False)
    person_skills = relationship("PersonSkill", back_populates="skill")

class PersonSkill(Base):
    __tablename__ = 'person_skill'
    __table_args__ = {'schema': 'backend'}
    id_person = Column(Integer, primary_key=True)
    id_skill = Column(Integer, ForeignKey('backend.skill.id_skill'), primary_key=True)
    yoe = Column(Integer)
    skill = relationship("Skill", back_populates="person_skills")

接下来是还原原PostgreSQL查询逻辑的代码：

# 输入参数：指定的技能列表和目标人员ID
input_skills = ['Spark', 'Oracle', 'Dataflow']
target_person_id = 1

# 1. 构建CTE：筛选指定技能并去重排序
skills_cte = (
    select(Skill.skill_name.distinct())
    .where(Skill.skill_name.in_(input_skills))
    .order_by(Skill.skill_name)
    .cte('skills')
)

# 2. 构建主查询，实现左连接和空值替换逻辑
query = (
    select(func.coalesce(PersonSkill.yoe, 0))
    .select_from(
        # 左连接CTE与skill表
        skills_cte.join(
            Skill,
            skills_cte.c.skill_name == Skill.skill_name,
            isouter=True
        )
        # 左连接skill表与person_skill表
        .join(
            PersonSkill,
            Skill.id_skill == PersonSkill.id_skill,
            isouter=True
        )
    )
    # 添加exists条件：过滤出目标人员存在技能记录的情况
    .where(
        exists(
            select(literal(1))
            .select_from(PersonSkill.join(Skill, PersonSkill.id_skill == Skill.id_skill))
            .where(PersonSkill.id_person == target_person_id)
        )
    )
    .order_by(Skill.skill_name)
)

# 执行查询（需提前创建好session对象）
result = session.execute(query).scalars().all()
# result为列表格式，对应每个输入技能的yoe值，无对应技能时返回0

关键细节说明： #

• 用.cte('skills')实现原SQL的WITH子句逻辑
• isouter=True标记左连接，确保即使目标人员无对应技能，该技能的记录也会被保留
• func.coalesce直接映射PostgreSQL的coalesce函数，处理yoe为空的场景返回0
• exists子查询完全还原原SQL的过滤逻辑

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