# SQLAlchemy-PGView

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

[![ty](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/ty/main/assets/badge/v0.json)](https://github.com/astral-sh/ty)
[![uv](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/uv/main/assets/badge/v0.json)](https://github.com/astral-sh/uv)
[![Ruff](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/ruff/main/assets/badge/v2.json)](https://github.com/astral-sh/ruff)
[![Python 3.10+](https://img.shields.io/badge/python-3.10+-blue.svg)](https://www.python.org/downloads/)
[![SQLAlchemy 2.0+](https://img.shields.io/badge/SQLAlchemy-2.0+-green.svg)](https://www.sqlalchemy.org/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

---

## 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

```python
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

```python
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:

```python
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:

```python
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

<div class="grid cards" markdown>

-   :material-book-open-variant:{ .lg .middle } **Views**

    ---

    Learn about regular views in depth

    [:octicons-arrow-right-24: Views](guide/views.md)

-   :material-database-refresh:{ .lg .middle } **Materialized Views**

    ---

    Caching, refreshing, and auto-refresh

    [:octicons-arrow-right-24: Materialized views](guide/materialized-views.md)

-   :material-source-branch:{ .lg .middle } **Alembic Migrations**

    ---

    Database migrations with autogenerate

    [:octicons-arrow-right-24: Alembic guide](guide/alembic.md)

-   :material-api:{ .lg .middle } **API Reference**

    ---

    Complete API documentation

    [:octicons-arrow-right-24: API reference](api/views/view.md)

</div>