5.5 KiB
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 12+
- Alembic 1.10+ (optional, for migrations)
Installation
=== "Base package"
bash uv pip install "sqlalchemy-pgview"
=== "With alembic support"
bash uv pip install "sqlalchemy-pgview[alembic]"
Quick Start
The recommended way to define views is using the declarative pattern with multiple inheritance. This integrates seamlessly with SQLAlchemy ORM models.
Define Your Models and Views
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 Views
from sqlalchemy import select
with engine.connect() as conn:
# Query regular view (always shows current data)
result = conn.execute(select(ActiveUsers.as_table())).fetchall()
for row in result:
print(f"{row.name}")
# Query materialized view (shows cached data)
stats = conn.execute(select(UserStats.as_table())).fetchall()
for stat in stats:
print(f"{stat.name}: {stat.order_count} orders, ${stat.total_spent}")
Refresh Materialized Views
Materialized views store cached results - refresh them when data changes:
with engine.begin() as conn:
UserStats.refresh(conn)
# Concurrent refresh (allows reads during refresh, requires unique index)
UserStats.refresh(conn, concurrently=True)
Auto-Refresh on Data Changes
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
Next Steps
-
:material-book-open-variant:{ .lg .middle } Views
Learn about regular views in depth
-
:material-database-refresh:{ .lg .middle } Materialized Views
Caching, refreshing, and auto-refresh
-
:material-source-branch:{ .lg .middle } Alembic Migrations
Database migrations with autogenerate
-
:material-api:{ .lg .middle } API Reference
Complete API documentation