Skip to content

Instantly share code, notes, and snippets.

@amcclosky
Last active July 24, 2020 17:25
Show Gist options
  • Save amcclosky/73e58ecf1982d7064adc57ba031ef5f4 to your computer and use it in GitHub Desktop.
Save amcclosky/73e58ecf1982d7064adc57ba031ef5f4 to your computer and use it in GitHub Desktop.
A global roughly time-ordered identifier mixin for sqlalchemy.

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.

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
)
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()
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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment