Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

make compatible with sqlalchemy #5

Merged
merged 3 commits into from
Oct 2, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 51 additions & 0 deletions examples/sqlalchemy.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
from typing import Optional

from sqlalchemy import types
from sqlalchemy.util import generic_repr
from typeid import TypeID, from_uuid


class TypeIDType(types.TypeDecorator):
"""
A SQLAlchemy TypeDecorator that allows storing TypeIDs in the database.
The prefix will not be persisted, instead the database-native UUID field will be used.
At retrieval time a TypeID will be constructed based on the configured prefix and the
UUID value from the database.

Usage:
# will result in TypeIDs such as "user_01h45ytscbebyvny4gc8cr8ma2"
id = mapped_column(
TypeIDType("user"),
primary_key=True,
default=lambda: TypeID("user")
)
"""
impl = types.Uuid

cache_ok = True

prefix: Optional[str]

def __init__(self, prefix: Optional[str], *args, **kwargs):
self.prefix = prefix
super().__init__(*args, **kwargs)

def __repr__(self) -> str:
# Customize __repr__ to ensure that auto-generated code e.g. from alembic includes
# the right __init__ params (otherwise by default prefix will be omitted because
# uuid.__init__ does not have such an argument).
return generic_repr(
self,
to_inspect=TypeID(self.prefix),
)

def process_bind_param(self, value: TypeID, dialect):
if self.prefix is None:
assert value.prefix is None
else:
assert value.prefix == self.prefix

return value.uuid

def process_result_value(self, value, dialect):
return from_uuid(value, self.prefix)
12 changes: 12 additions & 0 deletions tests/test_typeid.py
Original file line number Diff line number Diff line change
Expand Up @@ -81,3 +81,15 @@ def test_construct_type_from_uuid_with_prefix() -> None:
assert isinstance(typeid, TypeID)
assert typeid.prefix == "prefix"
assert isinstance(typeid.suffix, str)


def test_hash_type_id() -> None:
prefix = "plov"
suffix = "00041061050r3gg28a1c60t3gf"

typeid_1 = TypeID(prefix=prefix, suffix=suffix)
typeid_2 = TypeID(prefix=prefix, suffix=suffix)
typeid_3 = TypeID(suffix=suffix)

assert hash(typeid_1) == hash(typeid_2)
assert hash(typeid_3) != hash(typeid_1)
11 changes: 11 additions & 0 deletions typeid/typeid.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,10 @@ def suffix(self) -> str:
def prefix(self) -> str:
return self._prefix

@property
def uuid(self) -> UUID:
return _convert_b32_to_uuid(self.suffix)

def __str__(self) -> str:
value = ""
if self.prefix:
Expand All @@ -38,6 +42,9 @@ def __eq__(self, value: object) -> bool:
return False
return value.prefix == self.prefix and value.suffix == self.suffix

def __hash__(self) -> int:
return hash((self.prefix, self.suffix))


def from_string(string: str) -> TypeID:
prefix, suffix = get_prefix_and_suffix(string=string)
Expand Down Expand Up @@ -66,3 +73,7 @@ def get_prefix_and_suffix(string: str) -> tuple:

def _convert_uuid_to_b32(uuid_instance: UUID) -> str:
return base32.encode(list(uuid_instance.bytes))


def _convert_b32_to_uuid(b32: str) -> UUID:
return UUID(bytes=bytes(base32.decode(b32)))