af-db-types-json

Custom SQLAlchemy column types for PostgreSQL JSONB columns, backed by Pydantic models. Handles serialisation, deserialisation, and validation transparently — your columns read and write Pydantic model instances directly.

Installation

pip install af-db-types-json
# or with Poetry:
poetry add af-db-types-json

Types

JSONList[T] — JSONB array as a list of Pydantic models

Returns an empty list when the column value is NULL.

from pydantic import BaseModel
from sqlalchemy.orm import mapped_column, Mapped
from allfly.db.types.json import JSONList

class Tag(BaseModel):
    name: str
    colour: str

class Article(Base):
    __tablename__ = "articles"
    tags: Mapped[list[Tag]] = mapped_column(JSONList(Tag), default=list)

JSONObject[T] — JSONB object as a single Pydantic model

Returns None when the column value is NULL or validation fails (logs an error, does not raise).

from allfly.db.types.json import JSONObject

class Address(BaseModel):
    street: str
    city: str

class User(Base):
    __tablename__ = "users"
    address: Mapped[Address | None] = mapped_column(JSONObject(Address), nullable=True)

VersionedJSONObject[T] — JSONB object with automatic version-based dispatch

Stores a __type discriminator alongside your data so the correct model version is always deserialised, even as your schema evolves.

from allfly.db.types.json import VersionedJSONObject, VersionedModel

class PayloadV1(VersionedModel):
    version: float = 1.0
    data: str

class PayloadV2(VersionedModel):
    version: float = 2.0
    data: str
    extra: str

class Event(Base):
    __tablename__ = "events"
    payload: Mapped[PayloadV1 | PayloadV2 | None] = mapped_column(
        VersionedJSONObject(PayloadV1), nullable=True
    )

Rows written with PayloadV1 will deserialise as PayloadV1. Rows written with PayloadV2 will deserialise as PayloadV2. No migration needed.

VersionedModel — base class for versioned JSONB models

Subclasses must declare a version class attribute. The model_dump() method automatically injects a __type key (the class name) used for resolution.

from allfly.db.types.json import VersionedModel

class MyModelV1(VersionedModel):
    version: float = 1.0
    value: str

Configuring package scanning

By default VersionedModel.resolve_from_data() scans the src package (backwards compatible with the allfly-quest-core-api project). For any other project, call set_scan_packages once at app startup before any JSONB columns are read:

# In your app's startup / lifespan
from allfly.db.types.json import VersionedModel
import myapp

VersionedModel.set_scan_packages([myapp])

# Multiple packages are supported
VersionedModel.set_scan_packages([myapp, myapp_plugins])

# Disable scanning entirely (if you never use VersionedJSONObject)
VersionedModel.set_scan_packages([])

Logging

Validation errors in JSONObject and VersionedJSONObject are logged to the allfly.db.types.json.types logger namespace and return None rather than raising — this prevents a single bad row from crashing the application.

import logging
logging.getLogger("allfly.db.types.json").setLevel(logging.DEBUG)