51 lines
1.9 KiB
Python
51 lines
1.9 KiB
Python
|
|
|
|
from enum import Enum
|
|
from dataclasses import dataclass
|
|
from typing import Optional, Any
|
|
|
|
class DBErrorType(Enum):
|
|
"""Classified error categories."""
|
|
SUCCESS = "success"
|
|
NEED_MIGRATION = "need_migration"
|
|
MIGRATION_FAILED = "migration_failed"
|
|
OLD_CLIENT_NEW_API = "old_client_new_api"
|
|
MALFORMED_SQL = "malformed_sql_file"
|
|
MISSING_SQL = "missing_sql_file"
|
|
MISSING_DEPENDENCY = "missing_dependency"
|
|
INTEGRITY_ERROR = "integrity_error"
|
|
PERMISSION_ERROR = "permission_error"
|
|
FILESYSTEM_FULL = "filesystem_full"
|
|
DATABASE_LOCKED = "database_locked"
|
|
CORRUPTED_DATABASE = "corrupted_database"
|
|
PYTHON_MODEL_STRUCTURE = "python_model_structure"
|
|
UNKNOWN = "unknown"
|
|
|
|
@dataclass
|
|
class DatabaseOperation:
|
|
valid: bool
|
|
error_type: DBErrorType = DBErrorType.SUCCESS
|
|
data: Optional[Any] = None
|
|
message: Optional[str] = None
|
|
tried_migration: bool = False
|
|
tried_filtered: bool = False
|
|
|
|
def is_recoverable(self) -> bool:
|
|
"""Can the caller attempt a retry or manual fix?"""
|
|
return self.db_error_type in {
|
|
DBErrorType.OLD_CLIENT_NEW_API,
|
|
DBErrorType.NEED_MIGRATION
|
|
}
|
|
|
|
def user_message(self) -> str:
|
|
"""Human-readable error for the UI."""
|
|
messages = {
|
|
DBErrorType.SUCCESS: "Operation completed successfully.",
|
|
DBErrorType.NEED_MIGRATION: "Database schema needs update. Contact admin.",
|
|
DBErrorType.OLD_CLIENT_NEW_API: "Client version incompatible. Please upgrade.",
|
|
DBErrorType.MALFORMED_SQL: "Migration file is corrupted. Contact admin.",
|
|
DBErrorType.MISSING_SQL: "Migration file missing. Contact admin.",
|
|
DBErrorType.MISSING_DEPENDENCY: "Database dependency missing. Contact admin.",
|
|
DBErrorType.UNKNOWN: f"Unexpected error: {self.message}",
|
|
}
|
|
return messages.get(self.error_type, "Unknown error")
|