aegislm / backend /db /types.py
ACA050's picture
Upload 28 files
3e4aa3e verified
Raw
History Blame Contribute Delete
2.28 kB
"""
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:
# SQLite: Store as CHAR(36) string
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
# Handle string UUIDs
if isinstance(value, str):
value = uuid.UUID(value)
# Handle UUID objects
if not isinstance(value, uuid.UUID):
value = uuid.UUID(str(value))
if dialect.name == "postgresql":
return value
else:
# SQLite: Convert to string
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
# Handle string UUIDs from SQLite
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)
# Alias for convenience
GUIDType = GUID