1

我处于我想要访问的数据使用 SQL Server 中的表值函数的情况。

我正在构建的应用程序已经使用 SQLAlchemy ORM 和declarative_base(). 然后使用解析这些marshmallow_sqlalchemy.SQLAlchemyAutoSchema

当我尝试如下定义一个类时，我得到一个错误，表的 PK 无法映射，并且存储的列集合.c是一个空列表。

具体来说，错误是KeyError: 'PrimaryKey'
注释掉__mapper_args__结果：

sqlalchemy.exc.ArgumentError: Mapper mapped class TestModel->anon_1 could not assemble any primary key columns for mapped table 'anon_1'

import sqlalchemy
from sqlalchemy import create_engine, MetaData, func
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session

engine = create_engine("MY_CONN_STR")

# Create a locally scoped session object to be used for queries
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
db_session = scoped_session(SessionLocal)

class _Base:
    """
    Class defined to modify sqlalchemy declarative_base()
    """
    query = db_session.query_property()

Base = declarative_base(cls=_Base)


class TestModel(Base):
    """
    A Dynamic SQLAlchemy DB Model to use with a TVF
    """
    __table__ = func.mySchema.MyFuncWith2Defaults(None, None).table_valued()
    __table_args__ = {"schema": "mySchema"}
    __mapper_args__ = {"primary_key": func.mySchema.MyFuncWith2Defaults(None, None).table_valued().c["PrimaryKey"]}

    @classmethod
    def return_all_rows(cls) -> List["TestModel"]:
        return cls.query.all()

理想情况下，我想要达到的目的是将特定的 tvf 函数名称传递给函数，并让该函数为我创建类，如下所示：

def get_function_model(func_name: str, schema: str, pk: str=None) -> "TestModel":
    """
    Returns a SQLAlchemy DB Model of the supplied function name
    """
    # I'm aware the below will not work, but this is how I could do it with a normal Table I think
    attr_dict = {"__table__": metadata.tables[f"{schema}.{func_name}"]}
    if pk is not None:
        attr_dict["__mapper_args__"] = {"primary_key": attr_dict["__table__"].c[pk]}

    return type("TestModel", (TestModel,), attr_dict)

我可以使用 查询 TVF 精细engine.execute()，并在这样做时列标题按预期在行对象上拉出。

我的问题是，如何将 TVF 映射到 SQLAlchemy 类

4

0 回答 0