SQLAlchemy ORM左外连接：获取可导出CSV的合并查询结果问题

问题根源 #

你原代码的核心问题有两个：

1. 错误地将visit_id筛选条件放到了outerjoin的ON子句中，这和原SQL逻辑不符——原SQL是先完成所有左连接，再筛选visits.visit_id=1的记录，你的写法会导致连接逻辑和筛选条件绑定，可能返回不符合预期的结果。
2. 直接访问row.Patient.xxx时，未处理Patient为None的情况，触发AttributeError。

步骤1：修正查询语句 #

将筛选条件移到where子句，还原原SQL的逻辑：

def get_all_data_for_visit(self, visit_id: int):
    try:
        # 左连接仅关联patient_id，visit_id筛选放到where子句
        stmt = select(Visit, Patient)\
            .outerjoin(Patient, Visit.patient_id == Patient.patient_id)\
            .where(Visit.visit_id == visit_id)
        result = self.db.execute(stmt).all()
        return result if result else None
    except OperationalError as e:
        self.show_error_msg()

步骤2：安全遍历属性并导出CSV #

要保留SQL中的NULL值，需在访问属性前判断对象是否为None，可通过条件表达式或getattr实现，再将数据转换为CSV兼容格式。

手动指定字段的导出示例

import csv

def export_visit_data(visit_records, output_path):
    # 定义需要导出的所有字段（根据你的模型实际字段调整）
    fieldnames = [
        # Visit表字段
        "visit_id", "patient_id", "visit_type", "visit_date",
        # Patient表字段
        "patient_id", "full_name", "date_of_birth", "phone"
    ]

    with open(output_path, 'w', newline='', encoding='utf-8') as csv_file:
        writer = csv.DictWriter(csv_file, fieldnames=fieldnames)
        writer.writeheader()

        for visit, patient in visit_records:
            # 构建行数据，处理Patient为None的情况
            row_data = {
                # Visit对象不会为None（因为WHERE筛选的是visits表）
                "visit_id": visit.visit_id,
                "patient_id": visit.patient_id,
                "visit_type": visit.visit_type,
                "visit_date": visit.visit_date,
                # Patient对象可能为None，用条件表达式返回空值
                "patient_id": patient.patient_id if patient else "",
                "full_name": patient.full_name if patient else "",
                "date_of_birth": patient.date_of_birth if patient else "",
                "phone": patient.phone if patient else ""
            }
            writer.writerow(row_data)

通用字段获取（适配多表场景）

如果关联表较多，可通过SQLAlchemy的inspect自动获取模型字段，避免手动维护：

import csv
from sqlalchemy import inspect

def get_model_column_names(model_cls):
    """获取模型的所有字段名"""
    return [col.key for col in inspect(model_cls).columns]

def export_multitable_data(records, output_path):
    # 假设records是(Visit, Patient, OtherModel...)的元组
    visit_fields = get_model_column_names(Visit)
    patient_fields = get_model_column_names(Patient)
    # 如果有更多表，继续添加
    # other_fields = get_model_column_names(OtherModel)
    
    fieldnames = visit_fields + patient_fields
    # fieldnames += other_fields

    with open(output_path, 'w', newline='', encoding='utf-8') as csv_file:
        writer = csv.DictWriter(csv_file, fieldnames=fieldnames)
        writer.writeheader()

        for visit, patient in records:
            row = {}
            # 处理Visit字段
            for field in visit_fields:
                row[field] = getattr(visit, field)
            # 处理Patient字段
            for field in patient_fields:
                row[field] = getattr(patient, field) if patient else ""
            # 处理其他表字段
            # for field in other_fields:
            #     row[field] = getattr(other_model, field) if other_model else ""
            
            writer.writerow(row)

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