Can you map a table valued function to a declaritive base using SQLAlchemy

Question

I'm in a situation where the data I want to access is using Table Valued Functions in SQL Server.

The app I'm building already interfaces with regular tables using SQLAlchemy ORM and the declarative_base(). Those are then parsed using marshmallow_sqlalchemy.SQLAlchemyAutoSchema

When I try and define a class as below, I get an error that the PK of the table cannot be mapped, and the column collection stored in .c is an empty list.

Specifically the error is KeyError: 'PrimaryKey'
Commenting out the __mapper_args__ results in:

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()

Ideally the point I would like to get to is to pass a particular tvf function name to a function and have that function create the class for me, something like the below:

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)

I can query the TVF fine using engine.execute(), and in doing so the column headers pull through as expected on the row objects.

My question is, how can I map a TVF to a SQLAlchemy class

Can you map a table valued function to a declaritive base using SQLAlchemy

0 Answers0