SQLAlchemy from_select插入自定义值及列简写问题咨询

场景说明 #

现有两张分别存储员工信息、部门信息的数据表，需关联查询两张表数据后插入第三张新表，新表包含两张源表不存在的额外字段，已编写的基础代码如下：

stmt = select (
    employees.columns['emp_id'],
    employees.columns['f_name'],
    departments.columns['dept_id_dep'],
    departments.columns['dep_name']
)\
.select_from(
    employees.join(
        departments,
        employees.columns['dep_id'] == departments.columns['dept_id_dep'],
        isouter=True
        )
    )

EandP = Table('EmployeesPlusDepart', metadata,
       Column('Emp_id', String(50), primary_key = True, autoincrement = False), 
       Column('Name', String (50), index = False, nullable = False), 
       Column('Dept_id', String (50), nullable = False),
       Column('Dept_Name', String (50), nullable = False),
       Column('Location', String(50), default = 'CasaDuCarai', nullable = False), 
       Column('Start_date', Date,
           default = date.today() - timedelta(days=5), onupdate = date.today()),
       extend_existing=True, # 强制重定义metadata中的表配置
    )

Insert_stmt = insert(EandP).from_select(
    ['Emp_id', 'Name', 'Dept_id', 'Dept_Name'],
    stmt
)

当前待解决两个问题：

1. 执行插入时需要手动指定新表中Location、Start_date两个字段的值，如何在现有insert().from_select()语法中追加这类手动指定值
2. 编写select语句选取同一张表多个列时，是否支持类似employees.columns['emp_id','f_name']的简写方式简化代码

备注：Location、Start_date字段配置的默认值仅用于避免字段出现空值，无强制使用要求。

问题1解答：手动指定额外字段值的实现方式 #

from_select的逻辑是将SELECT查询返回的结果集按顺序映射到插入的目标字段，因此只需要把手动指定的值作为SQL字面量加入SELECT查询的字段列表，同时在插入字段列表中补上对应字段名即可，不需要修改其他逻辑。
具体操作：

• 导入SQLAlchemy的literal方法，用于将Python值转换为SQL层面的固定字面量
• 在原有SELECT语句的字段末尾，追加需要手动赋值的两个字段，用literal(自定义值)的形式传入，字段顺序可以自定义，只要和后续插入字段列表对应即可
• 在from_select的第一个参数（目标字段列表）末尾，按SELECT返回的字段顺序补上Location、Start_date两个字段名

修改后的核心代码示例：

from sqlalchemy import literal

# 追加手动指定的字段值到SELECT语句
stmt = select (
    employees.columns['emp_id'],
    employees.columns['f_name'],
    departments.columns['dept_id_dep'],
    departments.columns['dep_name'],
    literal("自定义Location值").label("Location"), # 替换为实际需要写入的Location值
    literal(date(2024, 5, 20)).label("Start_date") # 替换为实际需要写入的Start_date值
)\
.select_from(
    employees.join(
        departments,
        employees.columns['dep_id'] == departments.columns['dept_id_dep'],
        isouter=True
        )
    )

# 补充插入字段列表，和SELECT返回字段顺序严格对应
Insert_stmt = insert(EandP).from_select(
    ['Emp_id', 'Name', 'Dept_id', 'Dept_Name', 'Location', 'Start_date'],
    stmt
)

手动传入值时会自动覆盖字段配置的默认值，不需要额外调整默认值配置；如果某次插入不需要手动传值，保持原有写法即可，默认值会自动生效。

问题2解答：多列选择的简化写法 #

employees.columns['emp_id','f_name']这种传元组批量取列的写法不被支持，但是可以用以下几种官方支持的简写方式缩减代码长度：

• 用c作为columns的别名：employees.columns.emp_id可以简写为employees.c.emp_id，减少重复代码输入
• 批量取列时用列表推导：如果需要选取同一张表的多个指定列，可以把列名放到列表里，通过推导式批量获取，再解包传入select
• 选取全列时直接传表对象：如果需要查询某张表的所有列，直接在select中传入表对象即可，不需要逐一枚举列名

简化后的SELECT语句示例：

# 批量选取员工表、部门表需要的列
emp_cols = [employees.c[key] for key in ['emp_id', 'f_name']]
dept_cols = [departments.c[key] for key in ['dept_id_dep', 'dep_name']]

stmt = select(*emp_cols, *dept_cols)\
.select_from(
    employees.join(
        departments,
        employees.c.dep_id == departments.c.dept_id_dep,
        isouter=True
    )
)

内容的提问来源于stack exchange，提问作者FábioRB