58 lines
1.6 KiB
Python
58 lines
1.6 KiB
Python
import uuid
|
|
from sqlalchemy.dialects.postgresql import UUID
|
|
from sqlalchemy.dialects.sqlite import BLOB
|
|
from sqlalchemy.types import TypeDecorator, CHAR, Binary
|
|
|
|
|
|
class GUID(TypeDecorator):
|
|
"""Platform-independent GUID type.
|
|
|
|
Uses Postgresql's UUID type, otherwise uses
|
|
CHAR(32), storing as stringified hex values.
|
|
|
|
"""
|
|
impl = Binary
|
|
|
|
# if dialect.value == 'postgresql':
|
|
# impl = CHAR
|
|
# elif dialect.value == 'mysql':
|
|
# impl = MSBinary
|
|
# elif dialect.valie == 'sqlite':
|
|
# impl = Binary
|
|
# else:
|
|
# impl = Binary
|
|
|
|
def load_dialect_impl(self, dialect):
|
|
if dialect.name == 'postgresql':
|
|
return dialect.type_descriptor(UUID())
|
|
elif dialect.name == 'sqlite':
|
|
return dialect.type_descriptor(BLOB())
|
|
else:
|
|
return dialect.type_descriptor(CHAR(32))
|
|
|
|
def process_bind_param(self, value, dialect):
|
|
if value is None:
|
|
return None
|
|
elif dialect.name == 'postgresql':
|
|
return str(value)
|
|
elif not isinstance(value, uuid.UUID):
|
|
raise ValueError('value %s is not a valid uuid.UUID' % value)
|
|
else:
|
|
return value.bytes
|
|
# if not isinstance(value, uuid.UUID):
|
|
# return "%.32x" % uuid.UUID(value)
|
|
# else:
|
|
# # hexstring
|
|
# return "%.32x" % value
|
|
|
|
def process_result_value(self, value, dialect=None):
|
|
if value is None:
|
|
return None
|
|
elif isinstance(value, bytes):
|
|
return uuid.UUID(bytes=value)
|
|
else:
|
|
return uuid.UUID(value)
|
|
|
|
def is_mutable(self):
|
|
return False
|