Create a SQLModel class to query a MATERIALIZED VIEW

[Pk13055]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched the SQLModel documentation, with the integrated search.
• I already searched in Google "How to X in SQLModel" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to SQLModel but to Pydantic.
• I already checked if it is not related to SQLModel but to SQLAlchemy.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from datetime import date, datetime, time
from enum import Enum, unique
from typing import Optional, Tuple
from uuid import uuid4

from pydantic import UUID4, condecimal
from sqlalchemy import Column, DateTime
from sqlalchemy.dialects.postgresql import DATERANGE, ENUM, INT4RANGE, UUID
from sqlalchemy.schema import Index, PrimaryKeyConstraint
from sqlalchemy.sql import func
from sqlmodel import Field, SQLModel, create_engine, UniqueConstraint, select, Session, Relationship

from ..utils.db_view import create_materialized_view  # NOTE: Same/Similar to https://sqlalchemy-utils.readthedocs.io/en/latest/_modules/sqlalchemy_utils/view.html#create_materialized_view
from .user import User


@unique
class Season(str, Enum):
    """Season enum"""

    summer = "summer"
    winter = "winter"
    autumn = "autumn"
    spring = "spring"


@unique
class PeakType(str, Enum):
    """Peak usage? enum"""

    offpeak = "offpeak"
    partialpeak = "partialpeak"
    peak = "peak"


class PlanBase(SQLModel):
    """Base rate plan model"""

    id: Optional[int]
    uid: UUID4 = Field(
        default_factory=uuid4, nullable=False, sa_column=Column(UUID(as_uuid=True))
    )

    name: str
    is_deprecated: Optional[bool] = Field(default=False)


class Plan(PlanBase, table=True):
    """db plan model"""

    __table_args__ = (UniqueConstraint("uid"),)

    id: Optional[int] = Field(default=None, primary_key=True)

    added_on: datetime = Field(default_factory=datetime.utcnow)
    deprecated_on: Optional[datetime]

    rates: list["Rate"] = Relationship(back_populates="plan")
    users: list["User"] = Relationship(back_populates="plan")


class RateBase(SQLModel):
    """Plan rate model"""

    id: Optional[int]
    plan_id: Optional[int] = Field(foreign_key="plan.id")
    plan: Optional[Plan] = Relationship(back_populates="rates")

    rate: condecimal(max_digits=6, decimal_places=5) = Field(default=1.0)
    season: Season = Field(
        sa_column=Column(
            ENUM(Season, metadata=SQLModel.metadata),
            default=Season.summer,
        )
    )
    peak_type: PeakType = Field(
        sa_column=Column(
            ENUM(PeakType, metadata=SQLModel.metadata),
            default=PeakType.offpeak,
        )
    )

    date_range: Tuple[date, date] = Field(nullable=False, sa_column=Column(DATERANGE))
    time_range: Tuple[time, time] = Field(nullable=False, sa_column=Column(INT4RANGE))


class Rate(RateBase, table=True):
    """db rate model"""

    __table_args__ = (
        Index(
            "idx_plan_ts",
            "plan_id",
            "date_range",
            "time_range",
            postgresql_using="gist",
        ),
        UniqueConstraint("plan_id", "date_range", "time_range"),
    )

    id: Optional[int] = Field(default=None, primary_key=True)

    units: list["ConsumptionBase"] = Relationship(back_populates="rate")


class ConsumptionBase(SQLModel):
    """User consumption model"""

    user_id: int = Field(foreign_key="user.id")
    user: Optional["User"] = Relationship(back_populates="consumption")

    rate_id: Optional[int] = Field(foreign_key="rate.id")
    rate: Optional[Rate] = Relationship(back_populates="units")

    ts: datetime = Field(sa_column=Column(DateTime(timezone=True)))
    units: condecimal(max_digits=8, decimal_places=5) = Field(default=0.0)


class Consumption(ConsumptionBase, table=True):
    """db user consumption model"""

    __table_args__ = (
        PrimaryKeyConstraint("user_id", "ts"),
        UniqueConstraint("user_id", "ts"),
        {"timescaledb_hypertable": {"time_column_name": "ts"}},
    )


class Usage(SQLModel):
    """Materialized View combining user, rate and consumption data

    NOTE: Only for creation of view, **DO NOT** use this model for querying
    """

    # NOTE: Same/Similar to https://sqlalchemy-utils.readthedocs.io/en/latest/_modules/sqlalchemy_utils/view.html#create_materialized_view
    # from sqlalchemy_utils import create_materialized_view

    __table__ = create_materialized_view(
        name="usage",
        selectable=select(
            [
                Consumption.user_id,
                Consumption.ts,
                Rate.plan_id,
                Consumption.units,
                func.round(Rate.rate, 4).label("rate"),
                func.round(Consumption.units * Rate.rate, 2).label("cost"),
                Rate.season,
                Rate.peak_type,
            ]
        )
        .join(Rate, Rate.id == Consumption.rate_id)
        .order_by(Consumption.ts.desc()),
        metadata=SQLModel.metadata,
        indexes=[
            Index(
                "idx_usage_user_ts",
                "user_id",
                "ts",
                postgresql_using="gist",
                # postgresql_ops={ "ts": "DESC NULLS LAST"},  # NOTE: `gist` doesn't support `DESC NULLS LAST`
            ),
            Index(
                "idx_usage_user_plan_ts",
                "user_id",
                "plan_id",
                "ts",
                unique=True,
            ),
        ],
    )

engine = create_engine("timescaledb+postgresql://test:pass@localhost:5432/test")
SQLModel.metadata.create_all(engine)


with Session(engine) as session:
    query = select(Usage)
    usage = session.execute(query).scalars().all()  # TypeError: 'SQLModelMetaclass' object is not iterable

Description

• Create a materialized view
• Try to query it
• Since fields are not explicitly defined on the SQLModel object, the following TypeError is raised:

File "/usr/local/lib/python3.11/site-packages/sqlalchemy/sql/selectable.py", line 5166, in create_legacy_select
     self._raw_columns = [
                         ^
TypeError: 'SQLModelMetaclass' object is not iterable

Operating System

Linux, macOS, Other

Operating System Details

python3.11-alpine docker image running on Mac

SQLModel Version

0.0.6

Python Version

Python 3.11.4

Additional Context

I tried a few different approaches:

1. Create a new model UsageRead with the same __tablename__ and query it instead:

class UsageRead(SQLModel, table=True):
    __tablename__ = "usage"

    user_id: int
    ts: datetime
    plan_id: int
    units: condecimal(max_digits=8, decimal_places=5)
    rate: condecimal(max_digits=6, decimal_places=5)
    cost: condecimal(max_digits=8, decimal_places=2)
    season: Season
    peak_type: PeakType

This lead to a primary key error:

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

Trying any combination of table=True, PrimaryKeyConstraint in __table_args__ either returns an empty list or returns the results, but on subsequent restart of the backend service leads to recreation of the Usage view as a regular table.

(PS - you can find the same question posted on SO in case you want more information)

[Pk13055]

Solved it by using a custom class mapper! (ref.)

from datetime import datetime

from pydantic import condecimal
from sqlalchemy.orm import registry
from sqlalchemy.schema import Index
from sqlalchemy.sql import func
from sqlmodel import SQLModel, select

from ..utils.db_view import create_materialized_view

mapper_registry = registry()


class Usage(SQLModel):
    """View combining user, rate and consumption data

    NOTE: any changes to the `_Usage` schema must be reflected here
    """

    __tablename__ = "usage"

    user_id: int
    ts: datetime
    plan_id: int
    units: condecimal(max_digits=8, decimal_places=5)
    rate: condecimal(max_digits=6, decimal_places=5)
    cost: condecimal(max_digits=10, decimal_places=2)
    season: Season
    peak_type: PeakType


class _Usage(SQLModel):
    """View combining user, rate and consumption data

    NOTE: This is an internal model used **ONLY** for creating the materialized view
    """

    __table__ = create_materialized_view(
        name="usage",
        selectable=select(
            [
                Consumption.user_id,
                Consumption.ts,
                Rate.plan_id,
                Consumption.units,
                func.round(Rate.rate, 4).label("rate"),
                func.round(Consumption.units * Rate.rate, 2).label("cost"),
                Rate.season,
                Rate.peak_type,
            ]
        )
        .join(Rate, Rate.id == Consumption.rate_id)
        .order_by(Consumption.ts.desc()),
        metadata=SQLModel.metadata,
        indexes=[
            Index(
                "idx_usage_user_ts",
                "user_id",
                "ts",
                postgresql_using="gist",
                # postgresql_ops={ "ts": "DESC NULLS LAST"},  # NOTE: `gist` doesn't support `DESC NULLS LAST`
            ),
            Index(
                "idx_usage_user_plan_ts",
                "user_id",
                "ts",
                "plan_id",
                unique=True,
            ),
        ],
    )


mapper_registry.map_imperatively(Usage, _Usage.__table__)

Now I can use the Usage class as normal. Although this method requires me to maintain consistency between the fields of the view and the schema table, it is a small compromise.