pg-deltax 0.0.1

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 same shape as the local timescaledb package: SQLAlchemy engine/dialect helpers, SQLModel base models and sync 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

import deltax


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

    __tablename__ = "metrics"
    __partition_interval__ = "1 day"
    __premake__ = 3
    __enable_compression__ = True
    __segment_by__ = ["sensor_id"]
    __order_by__ = ["time"]
    __compression_policy_after__ = "7 days"
    __drop_after__ = "90 days"


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

DeltaxModel uses a composite primary key of id and time, matching the partitioning constraints PostgreSQL applies to time-partitioned tables.

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)


@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.