pg-deltax 0.0.2

Python helpers for the pg_deltax PostgreSQL extension with SQLModel, FastAPI, and Django support.

pg-deltax

pg-deltax is a Python companion package for the pg_deltax PostgreSQL extension. It keeps the Python layer thin: SQLAlchemy engine/dialect helpers, SQLModel table shapes, explicit wrappers for the pg_deltax SQL functions, FastAPI session glue, and a Django app/backend/migration layer.

DeltaX functions are schema-qualified as deltax.<function> to match the extension docs.

Install

pip install pg-deltax
pip install "pg-deltax[fastapi]"
pip install "pg-deltax[django]"

The database must have pg_deltax installed and loadable by PostgreSQL.

The distribution name is pg-deltax; the Python import package is deltax.

SQLModel

from sqlmodel import Field
from sqlmodel import Session
from sqlmodel import SQLModel

import deltax


class Metric(deltax.DeltaxModel, table=True):
    sensor_id: int = Field(index=True)
    value: float

    __tablename__ = "metrics"


engine = deltax.create_engine("postgresql+psycopg://user:pass@localhost/db")
SQLModel.metadata.create_all(engine)

with Session(engine) as session:
    deltax.activate_deltax_extension(session)
    deltax.create_deltatable(session, Metric, partition_interval="1 day", premake=3)
    deltax.enable_compression(
        session,
        Metric,
        segment_by=["sensor_id"],
        order_by=["time"],
    )
    deltax.set_compression_policy(session, Metric, "7 days")
    deltax.set_retention(session, Metric, "90 days")

DeltaxModel uses a composite primary key of id and time, matching the partitioning constraints PostgreSQL applies to time-partitioned tables. DeltaX settings live in explicit Deltatable specs or direct function calls, matching the pg_deltax extension API instead of SQLModel class attributes. For startup sync, a Deltatable spec runs the same pg_deltax setup from an explicit value:

metric_deltatable = deltax.Deltatable(
    Metric,
    partition_interval="1 day",
    premake=3,
    compression=deltax.CompressionConfig(
        segment_by=["sensor_id"],
        order_by=["time"],
        compress_after="7 days",
    ),
    drop_after="90 days",
)

deltax.create_all(engine, deltatables=[metric_deltatable])

Query helpers expose DeltaX analytics functions:

from sqlmodel import Session

with Session(engine) as session:
    rows = deltax.time_bucket_query(
        session,
        Metric,
        interval="1 hour",
        time_field="time",
        metric_field="value",
    )

FastAPI

from fastapi import Depends, FastAPI
from sqlmodel import Session

import deltax

engine = deltax.create_engine("postgresql+psycopg://user:pass@localhost/db")
get_session = deltax.fastapi.create_session_dependency(engine)

app = FastAPI()


@app.on_event("startup")
def startup() -> None:
    deltax.create_all(engine, deltatables=[metric_deltatable])


@app.get("/metrics")
def metrics(session: Session = Depends(get_session)):
    return deltax.time_bucket_query(session, Metric, metric_field="value")

Django

Add the app and use the backend if you want automatic extension creation:

INSTALLED_APPS = [
    "django.contrib.contenttypes",
    "deltax.django",
]

DATABASES = {
    "default": {
        "ENGINE": "deltax.django.db.backends.postgresql",
        "NAME": "app",
        "USER": "postgres",
        "PASSWORD": "postgres",
        "HOST": "localhost",
        "PORT": "5432",
        "OPTIONS": {"deltax_auto_create_extension": True},
    }
}

Define models with the Django shims:

from deltax.django.db import models


class Metric(models.DeltaxModel):
    time = models.DeltaxDateTimeField(interval="1 day", primary_key=True)
    sensor_id = models.IntegerField()
    value = models.FloatField()

Use migration operations to manage DeltaX:

from django.db import migrations
from deltax.django.db import migrations as deltax_migrations


class Migration(migrations.Migration):
    operations = [
        deltax_migrations.CreateExtension(),
        deltax_migrations.CreateDeltatable("Metric", time_column="time"),
        deltax_migrations.EnableCompression("Metric", segment_by=["sensor_id"]),
        deltax_migrations.SetCompressionPolicy("Metric", compress_after="7 days"),
        deltax_migrations.SetRetention("Metric", drop_after="90 days"),
    ]

Querysets can annotate with DeltaX functions:

Metric.objects.time_bucket("1 hour").values("bucket").annotate(avg=models.Avg("value"))

Live pg_deltax check

Use Docker Compose to build a PostgreSQL image with the upstream pg_deltax extension and run the live integration test:

docker compose up --build --abort-on-container-exit --exit-code-from live-test live-test

See docs/live-pg-deltax.md for details.