ULID primary keys as a Pydantic v2 custom type: working versions (tested 2026-10-15)

python-ulid==4.0.1
pydantic==2.14.1
pytest==9.1.1

import sqlite3
from typing import Annotated, Any

from pydantic import BaseModel, GetCoreSchemaHandler
from pydantic_core import core_schema
from ulid import ULID

CROCKFORD = r"[0-7][0-9A-HJKMNP-TV-Z]{25}"


def _validate(value: Any) -> ULID:
    if isinstance(value, ULID):
        return value
    if isinstance(value, str):
        # ULID.from_str() is case sensitive, so normalize first
        return ULID.from_str(value.strip().upper())
    if isinstance(value, int) and not isinstance(value, bool):
        return ULID.from_int(value)
    if isinstance(value, (bytes, bytearray)) and len(value) == 16:
        return ULID.from_bytes(bytes(value))
    raise ValueError("expected a ULID, a 26 char Crockford string, an int or 16 bytes")


class _UlidType:
    @classmethod
    def __get_pydantic_core_schema__(
        cls, source: Any, handler: GetCoreSchemaHandler
    ) -> core_schema.CoreSchema:
        return core_schema.no_info_plain_validator_function(
            _validate,
            serialization=core_schema.to_string_ser_schema(when_used="json-unless-none"),
            json_schema_input_schema=core_schema.str_schema(
                min_length=26, max_length=26, pattern=CROCKFORD
            ),
        )


Ulid = Annotated[ULID, _UlidType]


class User(BaseModel):
    id: Ulid
    name: str


class UserStore:
    def __init__(self) -> None:
        self.db = sqlite3.connect(":memory:")
        self.db.execute("CREATE TABLE users (id TEXT PRIMARY KEY, name TEXT NOT NULL)")

    def add(self, user: User) -> None:
        self.db.execute("INSERT INTO users VALUES (?, ?)", (str(user.id), user.name))

    def all(self) -> list[User]:
        rows = self.db.execute("SELECT id, name FROM users ORDER BY id").fetchall()
        return [User(id=row_id, name=name) for row_id, name in rows]

import sqlite3

import pytest
from pydantic import BaseModel, TypeAdapter, ValidationError
from ulid import ULID

from ulids import Ulid, User, UserStore


class PlainUser(BaseModel):
    id: ULID


def test_lowercase_string_is_normalized():
    uid = ULID()
    with pytest.raises(ValidationError):
        PlainUser(id=str(uid).lower())  # ULID.from_str rejects lowercase
    user = User(id=str(uid).lower(), name="ada")
    assert isinstance(user.id, ULID)
    assert user.id == uid
    assert str(user.id) == str(uid).upper()


def test_invalid_values_are_rejected():
    for bad in ("not-a-ulid", "0" + "I" * 25, 26 * "U", b"\x00" * 8):
        with pytest.raises(ValidationError):
            User(id=bad, name="x")


def test_json_round_trip_is_canonical_string():
    user = User(id=ULID(), name="ada")
    payload = user.model_dump_json()
    assert f'"id":"{user.id}"' in payload  # uppercase 26 char string
    assert User.model_validate_json(payload).id == user.id
    assert isinstance(user.model_dump()["id"], ULID)  # python mode keeps the rich type


def test_json_schema_is_a_plain_string():
    schema = TypeAdapter(Ulid).json_schema()
    assert schema["type"] == "string" and "anyOf" not in schema
    assert "anyOf" in TypeAdapter(ULID).json_schema()  # built in schema also allows 16 bytes


def test_ulids_order_chronologically_and_lexicographically():
    ids = [ULID.from_timestamp(seconds) for seconds in (2.0, 1.0, 3.0)]
    ordered = sorted(ids)
    assert [u.timestamp for u in ordered] == [1.0, 2.0, 3.0]
    assert [str(u) for u in ordered] == sorted(str(u) for u in ids)


def test_sqlite_primary_key_keeps_insertion_independent_order():
    store = UserStore()
    users = [User(id=ULID.from_timestamp(s), name=f"u{s}") for s in (3.0, 1.0, 2.0)]
    for user in users:
        store.add(user)
    assert [u.id.timestamp for u in store.all()] == [1.0, 2.0, 3.0]
    with pytest.raises(sqlite3.IntegrityError):
        store.add(users[0])

[pytest]
pythonpath = .