| """
|
| Custom SQLAlchemy Types for Cross-Dialect Compatibility
|
|
|
| Provides database-agnostic type definitions for:
|
| - GUID/UUID: Works with PostgreSQL (native UUID) and SQLite (CHAR(36))
|
| - This ensures the ORM is portable across different database backends.
|
| """
|
|
|
| import uuid
|
| from typing import Optional
|
|
|
| from sqlalchemy import CHAR, TypeDecorator
|
| from sqlalchemy.dialects.postgresql import UUID as PG_UUID
|
|
|
|
|
| class GUID(TypeDecorator):
|
| """
|
| Platform-independent GUID type.
|
|
|
| Uses PostgreSQL's native UUID type when available,
|
| otherwise uses CHAR(36) to store UUID values as strings.
|
|
|
| This ensures the ORM layer is database-agnostic and can
|
| work with both PostgreSQL and SQLite for testing purposes.
|
| """
|
|
|
| impl = CHAR
|
| cache_ok = True
|
|
|
| def load_dialect_impl(self, dialect):
|
| """Get the appropriate type descriptor for the dialect."""
|
| if dialect.name == "postgresql":
|
| return dialect.type_descriptor(PG_UUID(as_uuid=True))
|
| else:
|
|
|
| return dialect.type_descriptor(CHAR(36))
|
|
|
| def process_bind_param(self, value, dialect):
|
| """Convert Python UUID to database value."""
|
| if value is None:
|
| return value
|
|
|
|
|
| if isinstance(value, str):
|
| value = uuid.UUID(value)
|
|
|
|
|
| if not isinstance(value, uuid.UUID):
|
| value = uuid.UUID(str(value))
|
|
|
| if dialect.name == "postgresql":
|
| return value
|
| else:
|
|
|
| return str(value)
|
|
|
| def process_result_value(self, value, dialect):
|
| """Convert database value to Python UUID."""
|
| if value is None:
|
| return value
|
|
|
| if isinstance(value, uuid.UUID):
|
| return value
|
|
|
|
|
| return uuid.UUID(value)
|
|
|
| def compare_values(self, x, y):
|
| """Compare two values for equality."""
|
| if x is None or y is None:
|
| return x == y
|
| return str(x) == str(y)
|
|
|
|
|
|
|
| GUIDType = GUID
|
|
|