Robust Database Error Handling & Migrations System. For New Client upgrades (Operational Errors) & Old Clients with New JSON keys from an API (TypeErrors). We also added an SQL generic handling wrapper for use across any project or function.

core/controllers/ClientController.py

@@ -2,6 +2,8 @@
 from core.models.manage.session_management import init_session, close_session
 from core.services.sync.sync_service import coordinate_cache_sync, save_metadata
 from core.services.sync.orm_methods.sync_one_orm import sync_one_orm_model
+
+
 from core.errors.logger import logger
 
 # ORM models that can be sync'ed:
@@ -27,7 +29,13 @@ import pathlib
 import re
 import shutil
 import subprocess
+import time
 
+def evaluate_errors(final_result: dict, client_observer: Optional[ClientObserver] = None) -> str:
+    if not final_result["success"] and client_observer:
+        category = final_result["category"]
+        client_observer.notify('synchronizing', f'{category} Error! Check error logs')
+        time.sleep(2) # so they can see it.
 
 class ClientController:
 
@@ -93,6 +101,8 @@ class ClientController:
                 client_observer.notify('synchronizing', 'Fetching Locations List..')
 
             final_result = sync_one_orm_model(Location, "locations")
+            evaluate_errors(final_result)
+
 
         if "operators" in changed_tables:
             logger.info("Sync of Operators")
@@ -100,6 +110,7 @@ class ClientController:
                 client_observer.notify('synchronizing', 'Fetching Operators List..')
 
             final_result_two = sync_one_orm_model(Operator, "operators")
+            evaluate_errors(final_result_two)
 
         # =================== MANUAL-SQL BASED MODELS ==================
         try:

core/models/DatabaseOperation.py

@@ -0,0 +1,46 @@
+
+
+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"
+    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")

core/models/manage/insert.py

@@ -1,19 +1,27 @@
-from core.errors.logger import logger
+
 from sqlalchemy.orm import sessionmaker
 from sqlalchemy import create_engine
 from typing import Type, Dict, Any
 from sqlalchemy.exc import IntegrityError, SQLAlchemyError
+from sqlalchemy.orm import Session
+
+from core.errors.logger import logger
 from core.models.manage.session_management import get_session
 from core.models.orm_models.Base import BaseModel
 
+from core.models.manage.wrapper import safe_db_operation, WrapperRollback
+from core.models.DatabaseOperation import DatabaseOperation
+
 # all models it knows how to do:
 from core.models.orm_models.CachedSync import CachedSync
 from core.models.orm_models.Location import Location
 from core.models.orm_models.Operator import Operator
 from core.models.orm_models.EncryptedProxy import EncryptedProxy
 
-def insert_into_model(model_class: Type, all_data: dict | list, override=False) -> bool:
+# This is the public interface,
+def insert_into_model(model_class: Type, all_data: dict | list, override=False) -> DatabaseOperation:
     """
+    Purpose:
         Generic ORM insert for any model.  Keeping it generic for reuse.
     
     Args:
@@ -22,45 +30,86 @@ def insert_into_model(model_class: Type, all_data: dict | list, override=False)
         override: If True, wipe the table before inserting
     
     Returns:
-        True if successful
+        DatabaseOperation object with true/false
     """
-
     logger.info(f"All the public data preparing to be inserted is {all_data}")
 
-    with get_session() as session:
-        # Step 1: Wipe the table if override=True
-        if override:
-            logger.info(f"First, WIPING the pre-existing data for {model_class.__name__}. Are you sure you intended to completely delete the old data?")
-            session.query(model_class).delete()
-            session.commit()
-
-        logger.info(f"Starting insert for {model_class.__name__}")
-
     # Normalize to list for uniform handling
     data_list = all_data if isinstance(all_data, list) else [all_data]
 
+    # Call the wrapped function with normalized data
+    return _wrapped_insert(model_class=model_class, data_list=data_list, override=override)
+
+
+@safe_db_operation
+def _wrapped_insert(model_class: Type, data_list: dict | list, session: Session, override=False) -> DatabaseOperation:
+    """Insert items. Doesn't manage session."""
+
+    if override:
+        logger.info(f"First, WIPING the pre-existing data for {model_class.__name__}. Are you sure you intended to completely delete the old data?")
+        session.query(model_class).delete()
+
+    logger.info(f"Starting insert for {model_class.__name__}")
+
     try:
         for each_json in data_list:
             instance = model_class(**each_json)
             session.add(instance)
-            
+        session.commit()  # Commits both delete + inserts atomically
-            session.commit()
+        return DatabaseOperation(valid=True)
-            logger.info(f"Completed! SQL Insertion Successful following the {model_class.__name__} Model’s Type Rules")
-            return True
 
     except TypeError as e:
-            print(f"TypeError caught: {e}")
+        """
+        This is triggered when an Old client was sent unknown fields from the API in the JSON,
+        And it's handled inside the wrapped function because it needs access to the variables used.
+        """
         session.rollback()
-            raise ValueError(f"Invalid fields for {model_class.__name__}: {e}")
+
-        except IntegrityError as e:
+        valid_fields = {col.name for col in model_class.__table__.columns}
-            print(f"IntegrityError caught: {e.orig}")
+        filtered_data_list = [
-            session.rollback()
+            {k: v for k, v in each_json.items() if k in valid_fields}
-            raise ValueError(f"Constraint violation: {e.orig}")
+            for each_json in data_list
-        except SQLAlchemyError as e:
+        ]
-            print(f"SQLAlchemyError caught: {e}")
+        
-            session.rollback()
+        dropped_fields = set().union(*(set(d.keys()) - valid_fields for d in data_list))
-            raise
+        if dropped_fields:
-        except Exception as e:
+            logger.warning(f"Old Client Dropped unknown fields: {dropped_fields}")
-            print(f"Generic exception caught: {e}")
+        
-            session.rollback()
+        try:
-            raise e
+            for filtered_json in filtered_data_list:
+                instance = model_class(**filtered_json)
+                session.add(instance)
+            session.commit()
+            
+            # the fix worked,
+            return DatabaseOperation(
+                valid=True,
+                tried_filtered=True,
+                message=f"Inserted after filtering {dropped_fields}"
+            )
+        
+        except Exception as retry_error:
+            # Signal wrapper to rollback
+            raise WrapperRollback(DatabaseOperation(
+                valid=False,
+                error_type=DBErrorType.OLD_CLIENT_NEW_API,
+                tried_filtered=True,
+                message=str(f"Wrapped Insert Failed on Retry of a TypeError: {retry_error}")
+            ))
+
+
+# =================== Isolation Test ======================
+"""
+Spoof Failure of an operational error to trigger/test a migration
+"""
+    # from sqlalchemy import text
+
+    # # 1. Drop a column the model expects
+    # if model_class == Location:
+    #     with get_session() as session:
+    #         print("dropping provider_name from location..")
+    #         session.execute(text("ALTER TABLE locations DROP COLUMN provider_name"))
+    #         session.commit()
+    #         print("dropped provider_name from location!!!")
+
+    # # 2. Now insert will trigger OperationalError

core/models/manage/migrations.py

@@ -0,0 +1,93 @@
+# custom
+from core.errors.logger import logger
+from core.models.DatabaseOperation import DatabaseOperation, DBErrorType
+
+# generic
+from pathlib import Path
+from sqlalchemy import Column, Integer, String, create_engine, inspect, text
+from sqlalchemy.orm import declarative_base, Session
+from sqlalchemy.exc import OperationalError, DBAPIError
+from sqlalchemy import inspect
+
+"""
+Purpose:
+    Sync SQL database schema to current Python model definitions.
+
+Situation Used:
+    Python Models have new values missing from the same SQL database
+
+Metaphor:
+    This replaces something like alembic, by doing it directly
+
+Confusion:
+    This may be confusing because we're using SQLAlchemy,
+    but doing migrations manually.
+
+"""
+
+def migrate_sql() -> DatabaseOperation:
+    from core.models.manage.session_management import engine
+    from core.models.orm_models.Base import Base, BaseModel
+
+    try:
+        logger.info("[MIGRATION] Starting Migrations..")
+        inspector = inspect(engine)
+        existing_tables = inspector.get_table_names()
+        changes_made = []
+        
+        for table in Base.metadata.tables.values():
+            table_name = table.name
+
+            if table_name not in existing_tables:
+                table.create(engine)
+                added_table = f"Created table: {table_name}"
+                changes_made.append(added_table)
+                logger.info(f"[MIGRATION] {added_table}")
+                continue
+            
+            # Check for MISSING columns (add them)
+            existing_columns = {col['name'] for col in inspector.get_columns(table_name)}
+            model_columns = {col.name for col in table.columns}
+            missing_columns = model_columns - existing_columns
+            
+            if missing_columns:
+                with engine.begin() as conn:
+                    for col_name in missing_columns:
+                        col = table.columns[col_name]
+                        col_type = str(col.type.compile(engine.dialect))
+                        alter_sql = f"ALTER TABLE {table_name} ADD COLUMN {col_name} {col_type}"
+                        conn.execute(text(alter_sql))
+                
+                changes_made.append(f"Added columns to {table_name}: {missing_columns}")
+            
+            # Check for EXTRA columns (drop them)
+            extra_columns = existing_columns - model_columns
+            if extra_columns:
+                with engine.begin() as conn:
+                    for col_name in extra_columns:
+                        alter_sql = f"ALTER TABLE {table_name} DROP COLUMN {col_name}"
+                        logger.info(f"[MIGRATION] Dropping extra column: {alter_sql}")
+                        conn.execute(text(alter_sql))
+                
+                changes_made.append(f"Dropped extra columns from {table_name}: {extra_columns}")
+        
+        if changes_made:
+            logger.info(f"[MIGRATION] Completed! Schema synced successfully")
+            return DatabaseOperation(
+                valid=True,
+                data=changes_made,
+                message=f"Schema synced: {'; '.join(changes_made)}"
+            )
+        else:
+            return DatabaseOperation(
+                valid=True,
+                message="Schema already matches models"
+            )
+    
+    except (OperationalError, DBAPIError) as e:
+        logger.error(f"[MIGRATION] Failed: {str(e)}")
+        return DatabaseOperation(
+            valid=False,
+            error_type=DBErrorType.UNKNOWN,
+            message=f"Schema sync failed: {str(e)}"
+        )

core/models/manage/session_management.py

@@ -1,5 +1,8 @@
 from core.errors.logger import logger
 from core.models.orm_models.Base import BaseModel
+from core.models.manage.version_check import insert_new_version
+
+
 from sqlalchemy import create_engine, inspect
 from sqlalchemy.orm import sessionmaker
 import os
@@ -86,13 +89,38 @@ def create_ONLY_db_version_table():
         raise RuntimeError("Engine not initialized. Call _reinitialize_engine_and_session() first.")
     
     from core.models.orm_models.DatabaseVersion import database_version
-    try:
-        database_version.create(engine)
-    except:
-        logger.info("[DB MANAGEMENT] database_version already exists, but was attempted to be made again by calling create_ONLY_db_version_table.")
     
-    # if using without checking or try except blocks:
+    try:
-    # database_version.create(engine, checkfirst=True)
+        database_version.create(engine, checkfirst=True)
+        logger.info("[DB MANAGEMENT] database_version table created.")
+        
+        # CRITICAL FIX: Populate the table immediately
+        session = get_session()
+        existing_row = session.query(database_version).first()
+        
+        if not existing_row:
+            logger.info("[DB MANAGEMENT] Table was empty. Initializing with current app version.")
+            insert_new_version(session, Constants.DB_VERSION_THIS_APP_WANTS)
+        else:
+            logger.info("[DB MANAGEMENT] Table already has a version row.")
+            
+    except Exception as e:
+        logger.error(f"[DB MANAGEMENT] Failed to create/initialize database_version table: {e}")
+
+
+# def create_ONLY_db_version_table():
+#     """Create only the database_version table using the global engine."""
+#     if engine is None:
+#         raise RuntimeError("Engine not initialized. Call _reinitialize_engine_and_session() first.")
+    
+#     from core.models.orm_models.DatabaseVersion import database_version
+#     try:
+#         database_version.create(engine)
+#     except:
+#         logger.info("[DB MANAGEMENT] database_version already exists, but was attempted to be made again by calling create_ONLY_db_version_table.")
+
+#     # if using without checking or try except blocks:
+#     # database_version.create(engine, checkfirst=True)
 
 
 def does_db_version_table_exist():

core/models/manage/wrapper.py

@@ -0,0 +1,102 @@
+# custom
+from core.errors.logger import logger
+from core.models.DatabaseOperation import DatabaseOperation, DBErrorType
+from core.models.manage.migrations import migrate_sql
+from core.models.manage.session_management import get_session
+
+# generic
+from sqlalchemy.orm import Session
+from sqlalchemy.exc import (
+    OperationalError,
+    IntegrityError,
+    SQLAlchemyError
+)
+from typing import Type, Dict, Any, Callable, TypeVar
+from functools import wraps
+from enum import Enum
+
+
+def safe_db_operation(func):
+    def wrapper(*args, **kwargs):
+
+        # get the Session for the function it's wrapping,
+        session = get_session()
+
+        try:
+            # passes the session to the function,
+            result = func(*args, session=session, **kwargs)
+            
+            # If func already returns DatabaseOperation, pass it through
+            if isinstance(result, DatabaseOperation):
+                return result
+            
+            # Otherwise, wrap success
+            return DatabaseOperation(valid=True, data=result)
+        
+        except OperationalError as e:
+            logger.error(f"Schema mismatch in {func.__name__}: {e}")
+            session.rollback()
+            
+            try:
+                logger.info("Attempting migration...")
+                migrate_sql()
+
+                # try again:
+                result = func(*args, session=session, **kwargs)
+                return DatabaseOperation(
+                    valid=True,
+                    data=result,
+                    tried_migration=True,
+                    message="Recovered via migration"
+                )
+            except Exception as retry_error: # migration failed:
+                logger.error(f"Migration failed with error: {str(retry_error)}", exc_info=True)
+                session.rollback()
+                return DatabaseOperation(
+                    valid=False,
+                    error_type=DBErrorType.MIGRATION_FAILED,
+                    message=str(retry_error)
+                )
+
+        # This is really TypeErrors, passed from the insert function.
+        except WrapperRollback as e:
+            session.rollback()
+            return e.database_operation
+
+        except IntegrityError as e:
+            error_msg = f"IntegrityError caught: {e.orig}"
+            logger.error(error_msg)
+            session.rollback()
+            return DatabaseOperation(
+                valid=False,
+                error_type=DBErrorType.INTEGRITY_ERROR,
+                message=error_msg
+            )
+
+        except SQLAlchemyError as e:
+            error_msg = f"SQLAlchemyError caught: {e}"
+            logger.error(error_msg)
+            session.rollback()
+            return DatabaseOperation(
+                valid=False,
+                error_type=DBErrorType.UNKNOWN,
+                message=str(e)
+            )
+
+        except Exception as e:
+            # Catch-all for unexpected errors
+            session.rollback()
+            return DatabaseOperation(
+                valid=False,
+                error_type=DBErrorType.UNKNOWN,
+                message=str(e)
+            )
+    
+    return wrapper
+
+
+class WrapperRollback(Exception):
+    """Raised by wrapped function to signal wrapper should rollback and return error."""
+    def __init__(self, database_operation: DatabaseOperation):
+        self.database_operation = database_operation
+

core/services/sync/orm_methods/sync_one_orm.py

@@ -1,15 +1,20 @@
-# from core.services.sync.get_data_from_api import get_data_from_api
+# API
 from core.services.networking.api_requests.step1_get_or_post import get_data_from_api
 from core.services.networking.api_requests.ApiResponseModel import ApiResponse
 from core.services.networking.api_requests.step5_solve_api_problems import solve_api_problems
 
-
+# Database
+from core.models.DatabaseOperation import DatabaseOperation
 from core.models.manage.insert import insert_into_model
 from core.models.manage.denormalize import denormalize
 from core.models.orm_models.Base import Base
+
+# observers & loiggers
 from core.observers.ClientObserver import ClientObserver
 from core.observers.ConnectionObserver import ConnectionObserver
 from core.errors.logger import logger
+
+# basic utils
 from core.utils.basic_operations.get_parent_directory import get_parent_directory
 from core.Constants import Constants
 
@@ -41,6 +46,7 @@ def sync_one_orm_model(
         if not api_result.valid:
             return {
                 "success": False,
+                "category": "API",
                 "error": api_result.message
             }
 
@@ -57,25 +63,21 @@ def sync_one_orm_model(
         # Debug Pretty print with indentation
         # print(json.dumps(denormalized_data, indent=2))
 
-        try:
+        database_operation_object = insert_into_model(which_model, denormalized_data, True)
-            did_it_work = insert_into_model(which_model, denormalized_data, True)
+
-            if did_it_work:
+        if database_operation_object.valid:
             return {
                 "success": True
             }
         else:
             return {
                 "success": False,
-                    "error": f"We got the data from the endpoint {which_endpoint}, but failed to insert into the model.  That raw data is {denormalized_data}"
+                "category": "SQL",
+                "error": f"SQL Error: {database_operation_object.error_type}.  We got the data from the endpoint {which_endpoint}, but failed to insert into the model.  That raw data is {denormalized_data}"
             }
-        except:
-            return {
-                "success": False,
-                "error": f"The except triggered on a sync_one_orm_model's failure to insert into the model with the endpoint {which_endpoint}.  We got the raw data of {denormalized_data}"
-            }
-
     else:
         return {
             "success": False,
+            "category": "Data",
             "error": f"We had issues with denormalizing the data from the server.  That data was {new_data}."
         }