如何使用SQLAlchemy将SQL Server表值函数(TVF)映射到declarative_base类

错误原因 #

你遇到的报错根源有两点：

• 两次调用func.mySchema.MyFuncWith2Defaults(None, None).table_valued()会生成两个独立的匿名临时表对象，第二个对象没有加载列元数据，自然找不到PrimaryKey列抛出KeyError
• 直接使用table_valued()方法不会加载TVF的返回结构元数据，所以.c集合为空，ORM无法识别主键抛出ArgumentError

正确映射方式 #

映射TVF的核心逻辑是：先显式定义/反射TVF的返回表结构（指定主键），再将该表结构绑定到ORM映射类，查询时再传入参数调用TVF即可。

固定TVF映射示例 #

import sqlalchemy
from sqlalchemy import create_engine, MetaData, func, Table, Column, inspect
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session
from typing import List

engine = create_engine("MY_CONN_STR")

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
db_session = scoped_session(SessionLocal)

class _Base:
    query = db_session.query_property()

Base = declarative_base(cls=_Base)
# 反射TVF返回结构
inspector = inspect(engine)
tvf_columns = inspector.get_columns("MyFuncWith2Defaults", schema="mySchema")

# 构造带主键的Table对象
tvf_table = Table(
    "MyFuncWith2Defaults",
    Base.metadata,
    *[Column(c["name"], c["type"], primary_key=(c["name"] == "PrimaryKey")) for c in tvf_columns],
    schema="mySchema"
)

class TestModel(Base):
    __table__ = tvf_table

    @classmethod
    def return_all_rows(cls, param1=None, param2=None) -> List["TestModel"]:
        # 查询时传入参数调用TVF
        return cls.query.select_from(func.mySchema.MyFuncWith2Defaults(param1, param2)).all()

动态生成TVF映射类实现 #

和你预期的逻辑一致，封装为接收TVF名称、schema、主键名的通用函数即可：

from typing import Type

inspector = inspect(engine)

def get_function_model(func_name: str, schema: str, pk: str = None) -> Type[Base]:
    # 反射TVF返回列结构
    tvf_cols = inspector.get_columns(func_name, schema=schema)
    table_cols = []
    for col in tvf_cols:
        col_conf = {}
        if pk and col["name"] == pk:
            col_conf["primary_key"] = True
        table_cols.append(Column(col["name"], col["type"], **col_conf))
    
    # 构造Table对象
    tvf_table = Table(
        func_name,
        Base.metadata,
        *table_cols,
        schema=schema
    )
    # 动态生成ORM映射类
    return type(
        f"{func_name}Model",
        (Base,),
        {
            "__table__": tvf_table,
            "query_with_params": classmethod(
                lambda cls, *args: cls.query.select_from(
                    getattr(getattr(func, schema), func_name)(*args)
                ).all()
            )
        }
    )

使用示例 #

# 生成指定TVF的映射类
MyFuncModel = get_function_model("MyFuncWith2Defaults", schema="mySchema", pk="PrimaryKey")
# 带参数查询TVF
rows = MyFuncModel.query_with_params(None, None)
# 可直接适配marshmallow_sqlalchemy的SQLAlchemyAutoSchema使用，和平常ORM模型用法完全一致

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