如何在SQLAlchemy中查询多类型远程用户数据库时忽略不存在的列？

你的需求完全可以实现！这种对接用户自定义数据库的场景太常见了，核心思路就是先获取目标数据库表的实际列信息，再动态构建只包含这些列的查询语句，彻底避免引用数据库中不存在的字段。下面是具体的实现步骤：

方法一：手动筛选列并构建查询 #

利用SQLAlchemy的inspect工具可以直接读取数据库的元数据，对比你的模型列后筛选出存在的字段：

from sqlalchemy import inspect
from sqlalchemy.orm import Session

# 假设你的engine和Session已经初始化完成
session: Session = Session(bind=engine)

# 1. 获取数据库表的实际列名
inspector = inspect(engine)
# 替换成你的表名，这里是'contacts'
table_columns = inspector.get_columns('contacts')
existing_column_names = [col['name'] for col in table_columns]

# 2. 筛选模型中与数据库匹配的列
# 从Employee模型中提取存在的列对象
valid_columns = []
for col_name in Employee.__mapper__.columns.keys():
    if col_name in existing_column_names:
        valid_columns.append(getattr(Employee, col_name))

# 3. 只查询这些有效列
results = session.query(*valid_columns).all()

这样查询出来的结果只会包含数据库中实际存在的列，不会再抛出Unknown column的错误。如果需要将结果转为字典方便使用，可以这样处理：

result_dicts = [row._asdict() for row in results]

方法二：封装成工具函数复用 #

如果需要在多个模型上重复使用这个逻辑，可以把它封装成一个通用函数：

def get_model_valid_columns(model, engine):
    """获取模型中在数据库表中存在的列对象"""
    inspector = inspect(engine)
    table_name = model.__tablename__
    db_column_names = [col['name'] for col in inspector.get_columns(table_name)]
    return [getattr(model, col) for col in model.__mapper__.columns.keys() if col in db_column_names]

# 使用示例
valid_columns = get_model_valid_columns(Employee, engine)
results = session.query(*valid_columns).all()

注意事项 #

• 这个方法适用于单表查询，如果你的模型包含关联关系（比如外键关联其他表），需要额外处理关联表的列检查逻辑。
• 如果希望得到完整的Employee模型实例（包含不存在列的默认值），可以在查询后手动为实例的缺失字段赋值为None，不过通常直接使用字典结果会更灵活。

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