Given the following model schema definiton:
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
The materialized view and other data gets created sucessfully. Looking through the psql console, all indices, columns and subsequent constraints exist. Furthermore, I am able to sucessfully insert data into tables and have said dependent view show the updates. However, I want to be able to query the Usage data table, though because the fields/columns are not explicitly defined on the Usage model, I am unable to query them, specifically, facing this issue:
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
Inspite of trying different approache(s), viz creating 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
I am still facing (yet another) 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 - I have posted this same question in the sqlmodel discussion forum in case someone wants more information)
Finally solved it using this other SO answer. Basically, we need to create a dummy clone class with all the fields defined and map it to the View class.
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 Usage anywhere to query/filter/etc!