pygres

pygres is a small helper library around SQLAlchemy that focuses on:

• Declarative base and registry: a ready‑to‑use Base and registry for your models.
• PostgreSQL materialized views: a MaterializedView base class and helpers to create, refresh, and drop materialized views.
• Schema diffing: utilities to compare your SQLAlchemy models with an existing PostgreSQL database and render SQL statements to bring the schema in sync.

The goal is to keep the API minimal, explicit, and easy to reason about while staying close to plain SQLAlchemy.

---

Installation

Install via pip (or any PEP 621/pyproject-aware tool like uv, pip-tools, etc.):

pip install pygres

Or, in a uv‑managed project:

uv add pygres

You are expected to install and configure SQLAlchemy and a PostgreSQL driver (e.g. psycopg2) separately, as usual.

---

Quick start

1. Define models using the shared Base and registry

from sqlalchemy import Column, ForeignKey, Integer, String, Table
from sqlalchemy.orm import Mapped, mapped_column, relationship

from pygres import Base, registry


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


class AuditLog:
    pass


audit_logs_table = Table(
    "audit_logs",
    Base.metadata,
    Column("id", Integer, primary_key=True),
    Column("action", String(120), nullable=False),
)
registry.map_imperatively(AuditLog, audit_logs_table)

You can mix declarative mappings (via Base) and imperative mappings (via registry.map_imperatively) in the same metadata.

---

2. Define a materialized view

pygres provides a MaterializedView base class. You declare a materialized view by:

• Giving it a __tablename__
• Providing a SQLAlchemy select() query in __mv_query__
• Declaring the primary key column names in __mv_primary_key__

from sqlalchemy import select
from pygres import MaterializedView


class UserSummaryMV(MaterializedView):
    __tablename__ = "mv_user_summary"
    __mv_query__ = select(User.id.label("id"), User.name.label("name"))
    __mv_primary_key__ = ("id",)

The base class will:

• Create a Table with the right columns and types derived from the query.
• Mark the mapped table with info["is_materialized_view"] = True so that schema diffing and helpers know how to treat it.

You can control schema and “WITH DATA” behavior with:

• __mv_schema__: optional schema name.
• __mv_with_data__: True (default) or False to create the view WITH NO DATA.

---

3. Create and compare the schema against PostgreSQL

Below is the full example previously implemented in example.py, showing how to:

• Spin up a test PostgreSQL instance with testcontainers.
• Create tables and materialized views.
• Evolve the models.
• Compute and print SQL statements to migrate the live database schema.

from sqlalchemy import Column, ForeignKey, Integer, String, Table, create_engine, select
from sqlalchemy.orm import Mapped, mapped_column, relationship

from pygres import Base, registry, compare_database_schema, MaterializedView
from testcontainers.postgres import PostgresContainer


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


class AuditLog:
    pass


audit_logs_table = Table(
    "audit_logs",
    Base.metadata,
    Column("id", Integer, primary_key=True),
    Column("action", String(120), nullable=False),
)
registry.map_imperatively(AuditLog, audit_logs_table)


class UserSummaryMV(MaterializedView):
    __tablename__ = "mv_user_summary"
    __mv_query__ = select(User.id.label("id"), User.name.label("name"))
    __mv_primary_key__ = ("id",)


if __name__ == "__main__":
    # Start a temporary PostgreSQL instance
    with PostgresContainer("postgres:latest") as postgres:
        engine = create_engine(postgres.get_connection_url())

        # Create the initial schema (tables + materialized views) in the database
        registry.metadata.create_all(engine)

        # Evolve the models: add columns and a new table
        class UserV2(Base):
            __tablename__ = "users"
            id: Mapped[int] = mapped_column(primary_key=True)
            name: Mapped[str] = mapped_column(String(120))
            email: Mapped[str] = mapped_column(String(120))
            family_id: Mapped[int] = mapped_column(ForeignKey("families.id"))
            family: Mapped["Family"] = relationship("Family", back_populates="users")
            __table_args__ = {"extend_existing": True}

        class Family(Base):
            __tablename__ = "families"
            id: Mapped[int] = mapped_column(primary_key=True)
            name: Mapped[str] = mapped_column(String(120))
            users: Mapped[list[UserV2]] = relationship("UserV2", back_populates="family")

        # Compare metadata with the live database and get SQL migration statements
        sql_statements = compare_database_schema(
            engine,
            compare_server_default=False,
            return_sql=True,
        )
        for stmt in sql_statements:
            print(stmt)

Running this script will print SQL ALTER/CREATE/DROP statements that would bring the PostgreSQL schema in line with your current SQLAlchemy models and materialized views.

---

API overview

• pygres.Base: Declarative base for your models.
• pygres.registry: SQLAlchemy registry used to create Base and for imperative mappings.
• pygres.MaterializedView: Base class for PostgreSQL materialized view mappings.
• pygres.create_materialized_view(bind, model, if_not_exists=True): Execute and return the CREATE MATERIALIZED VIEW ... SQL for the given mapped view.
• pygres.refresh_materialized_view(bind, model, concurrently=False, with_data=True): Execute and return the REFRESH MATERIALIZED VIEW ... SQL.
• pygres.drop_materialized_view(bind, model, if_exists=True, cascade=False): Execute and return the DROP MATERIALIZED VIEW ... SQL.
• pygres.compare_database_schema(bind, **options):
  • With return_sql=False (default): returns a list of SchemaDiff objects.
  • With return_sql=True: returns a list of SQL strings that can be applied to migrate the schema.
• pygres.render_compensation_sql(diffs, dialect): Convert a list of SchemaDiff instances into SQL strings for a specific dialect.

All APIs are intentionally thin wrappers around SQLAlchemy primitives so that you can always drop down to plain SQLAlchemy when needed.

---

Notes

• Materialized view support currently targets PostgreSQL; attempting to use it with other dialects will raise a ValueError.
• compare_database_schema is designed for migration/compensation SQL generation, not as a full migration framework. You can integrate the generated SQL into your own deployment/migration tooling as you see fit.