A global roughly time-ordered identifier mixin for sqlalchemy.

README.md

A sqlalchemy model mixn, GlobalIdMixin which provides a global roughly time-ordered identifier and an obfuscated version of the global id for public consumption.

Makes use of python-ulid, PL/sql ULID and hashids-python.

hashid_util.py

import typing as t

import sys

from random import randint
from collections import namedtuple

from hashids import Hashids

DEFAULT_MIN_HASH_LENGTH = 12

PublicHash = namedtuple("PublicHash", "public_id salt")


def generate_hashid(
    ident: int, min_length: int = DEFAULT_MIN_HASH_LENGTH, salt: t.Optional[int] = None
) -> t.Tuple[str, int]:

    if salt is None:
        salt = randint(1, sys.maxsize)

    return PublicHash(
        Hashids(salt=str(salt), min_length=min_length).encode(ident), salt
    )

ulid_mixin.py

import typing as t

from datetime import date

import sqlalchemy as sa
import ulid

from sqlalchemy.dialects import postgresql
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.ext.indexable import index_property
from sqlalchemy.events import event
from sqlalchemy_utils import get_columns

from dateutil.relativedelta import relativedelta
from dateutil.rrule import rruleset, rrulestr
from dateutil.parser import parse as dt_parse

from pydantic import BaseModel

from ulid_type import ULIDType
from hashid_util import generate_hashid


class GlobalIdMixin(object):
    """
    Add an auto-generated, globally unique, time orderable id column
    based on ULID.
    """

    __func_schema__ = "util"

    @declared_attr
    def global_id(cls):
        return sa.Column(
            ULIDType(),
            nullable=True,
            unique=True,
            default=ulid.ULID,
            server_default=sa.text(f'("{cls.__func_schema__}".generate_ulid_uuid())'),
        )

    @declared_attr
    def public_id(cls):
        return sa.Column(sa.UnicodeText, nullable=True, unique=True)

    @declared_attr
    def public_id_salt(cls):
        return sa.Column(sa.BigInteger(), default=None, nullable=True)

    def init_public_id(self, salt: t.Optional[int] = None) -> None:
        if salt is None and self.public_id_salt is not None:
            salt = self.public_id_salt

        public_id, salt = generate_hashid(int(self.global_id), salt=salt)
        self.public_id = public_id
        self.public_id_salt = salt


def _global_id_objects(iter_):
    for obj in iter_:
        if isinstance(obj, GlobalIdMixin) and obj.public_id is None:
            yield obj


def enable_global_ids(session):
    @event.listens_for(session, "before_flush")
    def before_flush(session, flush_context, instances):
        for obj in _global_id_objects(session.dirty):
            obj.init_public_id()

        for obj in _global_id_objects(session.new):
            obj.init_public_id()

ulid_type.py

from __future__ import absolute_import
import uuid

import ulid

from sqlalchemy import types, util
from sqlalchemy.dialects import postgresql

from sqlalchemy_utils.types.scalar_coercible import ScalarCoercible
from ulid import ULID


class ULIDType(types.TypeDecorator, ScalarCoercible):
    """
    Stores a ULID in the database as a native UUID column type
    but can use TEXT if needed.
    ::
        from .lib.sqlalchemy_types import ULIDType
        class User(Base):
            __tablename__ = 'user'
            # Pass `force_text=True` to fallback TEXT instead of UUID column
            id = sa.Column(ULIDType(force_text=False), primary_key=True)
    """

    impl = postgresql.UUID(as_uuid=True)

    python_type = ulid.ULID

    def __init__(self, force_text=False, **kwargs):
        """
        :param force_text: Store ULID as TEXT instead of UUID.
        """
        self.force_text = force_text

    def __repr__(self):
        return util.generic_repr(self)

    def load_dialect_impl(self, dialect):
        if self.force_text:
            return dialect.type_descriptor(types.UnicodeText)

        return dialect.type_descriptor(self.impl)

    @staticmethod
    def _coerce(value):
        if not value:
            return None

        if isinstance(value, str):

            try:
                value = ulid.ULID.from_str(value)
            except (TypeError, ValueError):
                value = ulid.ULID.from_hex(value)

            return value

        if isinstance(value, uuid.UUID):
            return ulid.ULID.from_bytes(value.bytes)

        if not isinstance(value, ULID):
            return ulid.ULID.from_bytes(value)

        return value

    def process_bind_param(self, value, dialect):
        if value is None:
            return value

        if not isinstance(value, ulid.ULID):
            value = self._coerce(value)

        return str(value.to_uuid())

    def process_result_value(self, value, dialect):
        if value is None:
            return value

        return self._coerce(value)