sqlalchemy-pgview/README.md

SQLAlchemy-PGView

A SQLAlchemy 2.0+ extension that provides first-class support for PostgreSQL views and materialized views.

Features

• Declarative Views - Class-based view definitions using multiple inheritance with SQLAlchemy ORM
• View & MaterializedView Classes - Define PostgreSQL views as Python objects with full DDL support
• Alembic Integration - Database migration operations (op.create_view(), op.drop_view(), etc.)
• Auto-Refresh - Automatically refresh materialized views on data changes
• Async Support - Works with asyncpg and SQLAlchemy's async engines
• Dependency Tracking - Query PostgreSQL system catalogs for view dependencies
• Type Safety - Full type annotations for modern Python development

Requirements

• Python 3.10+
• SQLAlchemy 2.0+
• PostgreSQL database
• Alembic 1.10+ (optional, for migrations)

Installation

Base package

uv pip install "sqlalchemy-pgview"

With alembic support

uv pip install "sqlalchemy-pgview[alembic]"

Quick Start

from decimal import Decimal
from sqlalchemy import create_engine, select, func, String, Numeric, Integer
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
from sqlalchemy_pgview import ViewBase, MaterializedViewBase

# Define your base and models
class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    is_active: Mapped[bool] = mapped_column(default=True)

class Order(Base):
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(Integer)
    total: Mapped[Decimal] = mapped_column(Numeric(10, 2))

# Define a regular view (computed on every query)
class ActiveUsers(ViewBase, Base):
    __tablename__ = "active_users"
    __select__ = select(User.id, User.name).where(User.is_active == True)

# Define a materialized view (cached results, needs refresh)
class UserStats(MaterializedViewBase, Base):
    __tablename__ = "user_stats"
    __select__ = select(
        User.id.label("user_id"),
        User.name,
        func.count(Order.id).label("order_count"),
        func.coalesce(func.sum(Order.total), 0).label("total_spent"),
    ).select_from(User.__table__.outerjoin(Order.__table__, User.id == Order.user_id)
    ).group_by(User.id, User.name)

# Create everything (tables + views)
engine = create_engine("postgresql://user:pass@localhost/mydb")
Base.metadata.create_all(engine)

# Query the view
with engine.connect() as conn:
    result = conn.execute(select(ActiveUsers.as_table())).fetchall()
    for row in result:
        print(f"{row.name}")

# Refresh materialized view
with engine.begin() as conn:
    UserStats.refresh(conn)

Automatically refresh materialized views when underlying data changes:

from sqlalchemy.orm import Session

# Enable auto-refresh when Order table changes
UserStats.auto_refresh_on(Session, Order.__table__)

# Now commits automatically refresh the materialized view
with Session(engine) as session:
    session.add(Order(user_id=1, total=Decimal("100.00")))
    session.commit()  # UserStats is automatically refreshed

Alembic Integration

SQLAlchemy-PGView integrates with Alembic for automatic view detection and migration generation.

Import the alembic module in your env.py to enable autogenerate:

# env.py
import sqlalchemy_pgview.alembic  # Registers autogenerate support

Then generate migrations automatically:

alembic revision --autogenerate -m "add user views"

Alembic will detect:

• New views: Views in metadata but not in database
• Removed views: Views in database but not in metadata

You also can manually add refresh in existing migration:

def upgrade():
    # After data changes, refresh materialized views
    op.refresh_materialized_view("user_stats", concurrently=True)

API Reference

ViewBase (Declarative)

class MyView(ViewBase, Base):
    __tablename__ = "my_view"           # Required: view name
    __select__ = select(...)            # Required: SELECT statement
    __schema__ = "public"               # Optional: schema name

MaterializedViewBase (Declarative)

class MyMaterializedView(MaterializedViewBase, Base):
    __tablename__ = "my_mview"          # Required: view name
    __select__ = select(...)            # Required: SELECT statement
    __schema__ = "public"               # Optional: schema name
    __with_data__ = True                # Optional: populate on creation (default: True)

View (Imperative)

View(
    name: str,                          # View name
    selectable: Select,                 # SQLAlchemy SELECT statement
    schema: str | None = None,          # Schema name (default: public)
    metadata: MetaData | None = None,   # MetaData for auto-registration
)

MaterializedView (Imperative)

MaterializedView(
    name: str,                          # View name
    selectable: Select,                 # SQLAlchemy SELECT statement
    schema: str | None = None,          # Schema name (default: public)
    metadata: MetaData | None = None,   # MetaData for auto-registration
    with_data: bool = True,             # Populate on creation
)

License

MIT License - see LICENSE for details.

Contributing

Contributions are welcome! Please feel free to submit issues and pull requests.

Documentation

For full documentation, visit the docs directory.