sqlalchemy-pgview/docs/guide/views.md

Views

Views are virtual tables whose contents are defined by a query. Unlike regular tables, views don't store data - they compute results on each access.

Creating Views (Declarative)

The recommended way to create views is using the declarative pattern with multiple inheritance:

from sqlalchemy import select, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy_pgview import ViewBase

class Base(DeclarativeBase):
    pass

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

# Define a view using ViewBase
class ActiveUsers(ViewBase, Base):
    __tablename__ = "active_users"
    __select__ = select(User.id, User.name, User.email).where(User.is_active == True)

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

ViewBase Attributes

Attribute	Required	Description
__tablename__	Yes	Name of the view in the database
__select__	Yes	SELECT statement that defines the view
__schema__	No	Database schema (default: None/public)

Querying Views

from sqlalchemy import select

with engine.connect() as conn:
    # Query all rows
    result = conn.execute(select(ActiveUsers.as_table())).fetchall()

    # Query with filtering using .c accessor
    result = conn.execute(
        select(ActiveUsers.as_table())
        .where(ActiveUsers.c.name.like('A%'))
        .order_by(ActiveUsers.c.name)
    ).fetchall()

    for row in result:
        print(f"{row.name} ({row.email})")

Views with Joins

Views can encapsulate complex joins:

from sqlalchemy import func

class Author(Base):
    __tablename__ = "authors"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))

class Book(Base):
    __tablename__ = "books"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    author_id: Mapped[int] = mapped_column(Integer)
    price: Mapped[Decimal] = mapped_column(Numeric(10, 2))

# View with one-to-many join and aggregation
class AuthorStats(ViewBase, Base):
    __tablename__ = "author_stats"
    __select__ = select(
        Author.id,
        Author.name,
        func.count(Book.id).label("book_count"),
        func.avg(Book.price).label("avg_price"),
    ).select_from(
        Author.__table__.outerjoin(Book.__table__, Author.id == Book.author_id)
    ).group_by(Author.id, Author.name)

View Update Behavior

!!! info "Views Always Show Current Data" Regular views are just stored queries. When you INSERT, UPDATE, or DELETE rows in the underlying tables, the view immediately reflects those changes.

from sqlalchemy.orm import Session

with Session(engine) as session:
    # Add new user
    session.add(User(name="Alice", email="alice@example.com", is_active=True))
    session.commit()

# View immediately shows Alice
with engine.connect() as conn:
    result = conn.execute(select(ActiveUsers.as_table())).fetchall()
    # Alice is included!

Imperative API (Alternative)

For SQLAlchemy Core or when declarative style isn't suitable:

from sqlalchemy import MetaData, Table, Column, Integer, String, Boolean, select
from sqlalchemy_pgview import View

metadata = MetaData()

users = Table(
    "users", metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String(100)),
    Column("email", String(100)),
    Column("is_active", Boolean, default=True),
)

# Define the view
active_users = View(
    name="active_users",
    selectable=select(users.c.id, users.c.name, users.c.email)
        .where(users.c.is_active == True),
    metadata=metadata,
)

# Create tables and views
metadata.create_all(engine)

# Query the view
with engine.connect() as conn:
    result = conn.execute(select(active_users.as_table())).fetchall()

Dropping Views

Views are automatically dropped when using metadata.drop_all(). For manual control:

from sqlalchemy_pgview import DropView

with engine.begin() as conn:
    # Drop using view object
    conn.execute(DropView(ActiveUsers.as_view()))

    # Drop by name with options
    conn.execute(DropView("old_view", if_exists=True))

    # Drop with cascade (drops dependent objects)
    conn.execute(DropView("base_view", cascade=True))

List Registered Views

from sqlalchemy_pgview import get_views

# Get all registered views from metadata
views = get_views(Base.metadata)
print(views)  # {'active_users': <View 'active_users'>}

Complete Example

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

class Base(DeclarativeBase):
    pass

class Product(Base):
    __tablename__ = "products"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    category: Mapped[str] = mapped_column(String(50))
    price: Mapped[Decimal] = mapped_column(Numeric(10, 2))
    in_stock: Mapped[bool] = mapped_column(default=True)

# View for available products only
class AvailableProducts(ViewBase, Base):
    __tablename__ = "available_products"
    __select__ = select(
        Product.id,
        Product.name,
        Product.category,
        Product.price,
    ).where(Product.in_stock == True)

# View with aggregation by category
class CategoryPricing(ViewBase, Base):
    __tablename__ = "category_pricing"
    __select__ = select(
        Product.category,
        func.count(Product.id).label("product_count"),
        func.min(Product.price).label("min_price"),
        func.max(Product.price).label("max_price"),
        func.avg(Product.price).label("avg_price"),
    ).where(Product.in_stock == True).group_by(Product.category)

# Setup
engine = create_engine("postgresql://...")
Base.metadata.create_all(engine)

# Add products
with Session(engine) as session:
    session.add_all([
        Product(name="Widget", category="Tools", price=Decimal("9.99")),
        Product(name="Gadget", category="Electronics", price=Decimal("49.99")),
        Product(name="Doohickey", category="Tools", price=Decimal("19.99")),
    ])
    session.commit()

# Query views
with engine.connect() as conn:
    # Get available products
    available = conn.execute(select(AvailableProducts.as_table())).fetchall()
    print(f"Available products: {len(available)}")

    # Get category pricing
    pricing = conn.execute(
        select(CategoryPricing.as_table())
        .order_by(CategoryPricing.c.avg_price.desc())
    ).fetchall()

    for row in pricing:
        print(f"{row.category}: {row.product_count} products, avg ${row.avg_price:.2f}")