57 lines
1.5 KiB
Python
57 lines
1.5 KiB
Python
import uuid
|
|
from typing import Any
|
|
|
|
from sqlalchemy import Dialect
|
|
from sqlalchemy.dialects.postgresql import UUID
|
|
from sqlalchemy.types import CHAR, TypeDecorator
|
|
|
|
|
|
class GUID(TypeDecorator):
|
|
"""Platform-independent GUID type.
|
|
Uses PostgreSQL's UUID type, otherwise uses
|
|
CHAR(32), storing as stringified hex values.
|
|
"""
|
|
|
|
impl = CHAR
|
|
cache_ok = True
|
|
|
|
@staticmethod
|
|
def generate():
|
|
return uuid.uuid4()
|
|
|
|
@staticmethod
|
|
def convert_value_to_guid(value: Any, dialect: Dialect) -> str | None:
|
|
if value is None:
|
|
return value
|
|
elif dialect.name == "postgresql":
|
|
return str(value)
|
|
else:
|
|
if not isinstance(value, uuid.UUID):
|
|
return "%.32x" % uuid.UUID(value).int
|
|
else:
|
|
# hexstring
|
|
return "%.32x" % value.int
|
|
|
|
def load_dialect_impl(self, dialect):
|
|
if dialect.name == "postgresql":
|
|
return dialect.type_descriptor(UUID())
|
|
else:
|
|
return dialect.type_descriptor(CHAR(32))
|
|
|
|
def process_bind_param(self, value, dialect):
|
|
return self.convert_value_to_guid(value, dialect)
|
|
|
|
def _uuid_value(self, value):
|
|
if value is None:
|
|
return value
|
|
else:
|
|
if not isinstance(value, uuid.UUID):
|
|
value = uuid.UUID(value)
|
|
return value
|
|
|
|
def process_result_value(self, value, dialect):
|
|
return self._uuid_value(value)
|
|
|
|
def sort_key_function(self, value):
|
|
return self._uuid_value(value)
|