Plugin architecture

pgcraft’s resource factories are built entirely on a plugin system. A factory is a thin runner that topologically sorts a list of plugins by their declared dependencies and calls each one in turn. All behaviour — primary keys, table layout, views, triggers, API exposure — is provided by plugins, so every part of it can be replaced, extended, or recomposed without touching the core.

Core concepts

Plugin

A plugin is any instance of a class that inherits from Plugin. The run() method is a no-op by default, so a plugin only needs to implement the hooks it cares about. Plugins communicate by writing to and reading from a shared FactoryContext.

FactoryContext

FactoryContext carries the inputs supplied to the factory (table name, schema, metadata, schema items) together with a flat key/value store that plugins use to pass objects between themselves.

# Writing a value
ctx["my_table"] = table

# Reading a value
table = ctx["my_table"]

# Checking existence (safe; never raises)
if "my_table" in ctx:
    ...

# Intentional override (force=True required to prevent accidents)
ctx.set("my_table", replacement, force=True)

Writing to a key that already exists raises KeyError immediately. Reading a key that has not yet been written raises KeyError with a hint naming the missing key.

ResourceFactory

ResourceFactory is the plugin runner. It takes a list of plugins, validates it for singleton conflicts, topologically sorts the plugins by their produces() / requires() declarations, and calls run() on each in the resolved order.

Subclasses declare DEFAULT_PLUGINS to establish their standard behaviour. The three built-in dimension factories are thin wrappers:

class PGCraftSimple(ResourceFactory):
    DEFAULT_PLUGINS = []

    _INTERNAL_PLUGINS = [
        SimpleTablePlugin(),
        TableCheckPlugin(),
        RawTableProtectionPlugin("primary"),
    ]

Plugin execution order

Execution order is determined by sorting plugins topologically using two class decorators: produces() and requires().

produces()

Declare the ctx keys this plugin’s run() method will write.

requires()

Declare the ctx keys this plugin’s run() method needs to already be set before it runs.

The factory builds a dependency graph from these declarations and calls run() in a valid topological order. Plugins with no relationship to each other preserve their original list order.

from pgcraft.plugin import Dynamic, Plugin, produces, requires

@produces(Dynamic("out_key"))
@requires("primary")
class MyTransformPlugin(Plugin):
    """Derive a summary table from the primary table."""

    def __init__(self, out_key: str = "summary") -> None:
        self.out_key = out_key

    def run(self, ctx: FactoryContext) -> None:
        primary = ctx["primary"]
        # ... build summary from primary ...
        ctx[self.out_key] = summary

MyTransformPlugin will always run after the plugin that produces "primary", regardless of the order they appear in the plugin list.

Injected columns

Some plugins need to contribute columns to a table without knowing which table plugin will consume them. For this, plugins append Column objects to ctx.injected_columns — a shared list on FactoryContext. Table plugins that support this pattern (currently LedgerTablePlugin) spread the list into the table definition:

# Column-providing plugin
class MyColumnPlugin(Plugin):
    def run(self, ctx: FactoryContext) -> None:
        ctx.injected_columns.append(
            Column("tenant_id", Integer, nullable=False)
        )

# Table plugin spreads injected columns
table = Table(
    ctx.tablename, ctx.metadata,
    *pk_columns,
    *ctx.injected_columns,   # entry_id, created_at, etc.
    Column("value", Integer(), nullable=False),
    *ctx.table_items,
    schema=ctx.schemaname,
)

Built-in plugins that inject columns:

• CreatedAtPlugin — injects a created_at timestamp column.

• UUIDEntryIDPlugin — injects an entry_id UUID column.

• DoubleEntryPlugin — injects a direction column for debit/credit semantics.

These plugins also write their standard ctx store keys (e.g. "created_at_column", "entry_id_column", "double_entry_columns") so that downstream trigger plugins can look up column metadata.

Before any run() call the factory collects two special inputs:

pk_columns

The first non-None result across all plugins is stored in ctx.pk_columns.

extra_columns

Results from all plugins are concatenated and stored in ctx.extra_columns.

Both are available to every plugin’s run() via the typed fields ctx.pk_columns and ctx.extra_columns.

Dynamic key references

When a ctx key name is a constructor parameter rather than a fixed string, use Dynamic inside the decorator:

@produces(Dynamic("table_key"))
class SimpleTablePlugin(Plugin):
    def __init__(self, table_key: str = "primary") -> None:
        self.table_key = table_key

    def run(self, ctx: FactoryContext) -> None:
        ctx[self.table_key] = Table(...)

The decorator validates at class definition time that the Dynamic attribute name is a real __init__ parameter, so typos are caught immediately.

PostgreSQL version requirements

Plugins that rely on features introduced in a specific PostgreSQL version can declare this via MinPGVersion inside the requires() decorator:

from pgcraft.plugin import MinPGVersion, Plugin, produces, requires

@requires(MinPGVersion(18))
@produces("pk_columns")
class UUIDV7PKPlugin(Plugin):
    """Requires PostgreSQL 18+ for uuidv7()."""
    ...

MinPGVersion is filtered out of the ctx dependency graph — it only sets min_pg_version on the class. To validate at runtime, call check_pg_version() with the connected server’s major version:

from pgcraft.plugin import check_pg_version

with engine.connect() as conn:
    major = conn.dialect.server_version_info[0]
    check_pg_version(major, factory.ctx.plugins)

This raises PGCraftValidationError with a clear message instead of letting PostgreSQL fail with “function does not exist”.

Plugin resolution

Each factory call resolves its plugin list from three sources, concatenated in this order:

1. Global plugins — from PGCraftConfig, if passed via the config= argument. Prepended before everything else.

2. Factory plugins — plugins kwarg if supplied, otherwise DEFAULT_PLUGINS.

3. Extra plugins — always appended via extra_plugins.

pgcraft_cfg = PGCraftConfig()
pgcraft_cfg.register(AuditPlugin())         # prepended to every factory

PGCraftSimple(
    "users", "app", metadata, schema_items,
    config=pgcraft_cfg,                   # global plugins first
    extra_plugins=[TenantPlugin()],       # appended after defaults
)

To replace the default plugin list entirely:

events = PGCraftSimple(
    "events", "app", metadata, schema_items,
    plugins=[                             # replaces DEFAULT_PLUGINS
        UUIDV4PKPlugin(),
    ],
)

Singleton groups

Some plugins must appear at most once in a resolved list (e.g. you cannot have two PK plugins). The singleton() decorator declares a group name; the factory raises PGCraftValidationError at construction time if two plugins share the same group.

from pgcraft.plugin import Plugin, singleton

@singleton("__pk__")
class MyPKPlugin(Plugin):
    ...

The built-in groups are "__pk__" (one PK plugin), "__table__" (one table-layout plugin), "__entry_id__" (one entry ID plugin), and "__double_entry__" (one double-entry plugin). You can define your own group names for custom plugins.

Context keys

Plugins read and write objects in ctx using string keys. Every built-in plugin accepts its key names as constructor arguments with sensible defaults, so two independent pipelines can coexist in one factory without colliding.

SerialPKPlugin / UUIDV4PKPlugin / UUIDV7PKPlugin

Writes "pk_columns". UUIDV7PKPlugin additionally declares MinPGVersion(18).

SimpleTablePlugin

Writes "primary" (the backing table).

SimpleTriggerPlugin

Reads "primary" (via table_key) and "api" (via view_key). Accepts columns (writable column subset) and permitted_operations (which DML operations get INSTEAD OF triggers).

RawTableProtectionPlugin

Reads the table keys passed to its constructor (e.g. "primary", "root_table", "attributes"). Installs BEFORE triggers that block direct DML on raw backing tables — mutations must go through the API view. Included automatically in each factory’s _INTERNAL_PLUGINS.

AppendOnlyTablePlugin

Writes "root_table" and "attributes".

AppendOnlyViewPlugin

Reads "root_table" and "attributes". Writes "primary".

AppendOnlyTriggerPlugin

Reads "root_table", "attributes", and "api" (optional; skipped if absent from ctx).

EAVTablePlugin

Writes "entity", "attribute", and "eav_mappings".

EAVViewPlugin

Reads "entity", "attribute", and "eav_mappings". Writes "primary".

EAVTriggerPlugin

Reads "entity", "attribute", "eav_mappings", and "api" (optional; skipped if absent from ctx).

UUIDEntryIDPlugin

Writes "entry_id_column" (a Column object). Also appends the column to ctx.injected_columns.

CreatedAtPlugin

Writes "created_at_column" (the column name string). Also appends a DateTime column to ctx.injected_columns.

LedgerTablePlugin

Reads "pk_columns" and ctx.injected_columns. Requires "entry_id_column" and "created_at_column" for ordering. Writes "primary" (the table) and "__root__".

LedgerTriggerPlugin

Reads "primary" (via table_key), "api" (via view_key), and "entry_id_column".

LedgerLatestViewPlugin

Reads "primary" (via table_key) and "created_at_column". Writes "latest_view" (via latest_view_key).

LedgerBalanceViewPlugin

Reads "primary" (via table_key). Writes "balance_view" (via balance_view_key).

LedgerBalanceCheckPlugin

Reads "primary" (via table_key). Registers an AFTER INSERT trigger enforcing SUM(value) >= min_balance per dimension group.

DoubleEntryPlugin

Writes "double_entry_columns" (the direction column name). Appends a direction column to ctx.injected_columns.

DoubleEntryTriggerPlugin

Reads "primary" (via table_key), "double_entry_columns", and "entry_id_column".

All key names are overridable via constructor arguments, which means you can wire plugins together in non-standard ways or run multiple pipelines within a single factory.

Writing a custom plugin

Implement run() and declare your dependencies with produces() and requires(). Use ctx to pass objects to downstream plugins.

A simple plugin that only contributes extra columns needs no dependency declarations at all — it implements extra_columns() instead of run():

from __future__ import annotations

from sqlalchemy import Column, DateTime, func
from pgcraft.plugin import Plugin
from pgcraft.factory.context import FactoryContext


class TimestampPlugin(Plugin):
    """Add ``created_at`` / ``updated_at`` columns to every table."""

    def extra_columns(self, _ctx: FactoryContext) -> list[Column]:
        return [
            Column("created_at", DateTime(timezone=True),
                   server_default=func.now()),
            Column("updated_at", DateTime(timezone=True),
                   server_default=func.now(), onupdate=func.now()),
        ]

Register it globally so it applies to every factory in the project:

from pgcraft.config import PGCraftConfig
from pgcraft.factory import PGCraftAppendOnly, PGCraftSimple

pgcraft_cfg = PGCraftConfig()
pgcraft_cfg.register(TimestampPlugin())

PGCraftSimple(
    "products", "app", metadata, schema_items,
    config=pgcraft_cfg,
)
PGCraftAppendOnly(
    "orders", "app", metadata, schema_items,
    config=pgcraft_cfg,
)

Or apply it to a single factory only:

PGCraftSimple(
    "products", "app", metadata, schema_items,
    extra_plugins=[TimestampPlugin()],
)

Custom plugin with ctx communication

Here is a more involved example: a plugin that creates a shadow audit table and makes it available to a downstream trigger plugin via a ctx key.

from sqlalchemy import Column, DateTime, ForeignKey, Integer, Table
from pgcraft.plugin import Dynamic, Plugin, produces, requires, singleton
from pgcraft.factory.context import FactoryContext


@produces(Dynamic("shadow_key"))
@singleton("__shadow__")
class ShadowTablePlugin(Plugin):
    """Create a shadow audit table alongside the main table."""

    def __init__(self, shadow_key: str = "shadow") -> None:
        self.shadow_key = shadow_key

    def run(self, ctx: FactoryContext) -> None:
        shadow = Table(
            f"{ctx.tablename}_shadow",
            ctx.metadata,
            Column("id", Integer, primary_key=True),
            Column("ref_id", Integer,
                   ForeignKey(f"{ctx.schemaname}.{ctx.tablename}.id")),
            Column("changed_at", DateTime(timezone=True)),
            schema=ctx.schemaname,
        )
        ctx[self.shadow_key] = shadow


@requires(Dynamic("shadow_key"))
class ShadowTriggerPlugin(Plugin):
    """Register a trigger that writes to the shadow table on every change."""

    def __init__(self, shadow_key: str = "shadow") -> None:
        self.shadow_key = shadow_key

    def run(self, ctx: FactoryContext) -> None:
        shadow = ctx[self.shadow_key]   # guaranteed by @requires
        # ... register trigger using shadow.name, etc.


# Use them together — order in the list doesn't matter because the
# dependency declarations ensure ShadowTablePlugin runs first.
PGCraftSimple(
    "products", "app", metadata, schema_items,
    extra_plugins=[ShadowTriggerPlugin(), ShadowTablePlugin()],
)

Global configuration

PGCraftConfig holds the global plugin list that is prepended to every factory that references it.

from pgcraft.config import PGCraftConfig

pgcraft_cfg = PGCraftConfig()
pgcraft_cfg.register(TimestampPlugin(), TenantPlugin())

# -- or equivalently --
pgcraft_cfg = PGCraftConfig(plugins=[TimestampPlugin(), TenantPlugin()])

Pass it to each factory via the config= argument. A common pattern is to create one PGCraftConfig per project and import it wherever factories are defined:

# pgcraft_setup.py
from pgcraft.config import PGCraftConfig
from myapp.plugins import TimestampPlugin, TenantPlugin

pgcraft_cfg = PGCraftConfig()
pgcraft_cfg.register(TimestampPlugin(), TenantPlugin())

# models.py
from myapp.pgcraft_setup import pgcraft_cfg
from pgcraft.factory import PGCraftSimple

PGCraftSimple(
    "users", "app", metadata, schema_items,
    config=pgcraft_cfg,
)

Factory and view reference

Table factories create the core data model. Each factory type has its own internal plugins and trigger strategy:

PGCraftSimple

Single backing table. Best for reference data and simple lookups.

PGCraftAppendOnly

SCD Type 2 dimension with root + attributes tables and a current-state join view. Best for slowly changing dimensions where audit trails matter.

PGCraftEAV

Entity-Attribute-Value dimension with entity + attribute tables and a pivot view. Best for sparse or highly dynamic attributes.

PGCraftLedger

Append-only ledger table with value, entry_id, and created_at columns. Best for event logs, financial journals, and metric observations.

View factories create derived views from a table factory:

PostgRESTView

PostgREST-facing view with auto-selected INSTEAD OF triggers. Grants drive which triggers are created. Supports columns, exclude_columns, and query= for customisation.

PGCraftView

Standalone view from any SQLAlchemy select() expression. Exposes .table for use in joins.

PGCraftMaterializedView

Materialized view with an auto-generated refresh function.

BalanceView

Ledger balance view (SUM(value) GROUP BY dimensions).

LatestView

Ledger latest view (DISTINCT ON (dimensions) ORDER BY created_at DESC).

LedgerActions

Ledger event functions for reconciliation and delta inserts.

Built-in plugins reference

See Built-in plugins for detailed documentation of every built-in plugin, including parameters, context keys, and usage examples.

See Extension system for the extension system that sits above plugins, bundling plugins with metadata hooks, Alembic hooks, and CLI commands.