diff --git a/core/models/manage/clear_sql_model.py b/core/models/manage/clear_sql_model.py new file mode 100644 index 0000000..08de9d2 --- /dev/null +++ b/core/models/manage/clear_sql_model.py @@ -0,0 +1,97 @@ + +from core.models.DatabaseOperation import DatabaseOperation, DBErrorType +from core.models.manage.wrapper import safe_db_operation +from core.errors.logger import logger + +from typing import Type +from sqlalchemy.orm import Session + + + +@safe_db_operation +def _drop_sql_model_wrapped(model_class: Type, session=None) -> DatabaseOperation: + """ + Purpose: + Drop the entire table + + Confusion: + It gets the session from the wrapper + The wrapper converts the returning dictionary into DatabaseOperation.data + """ + model_class.__table__.drop(bind=session.bind) + session.commit() + return { + 'dropped': model_class.__name__, + 'action': 'drop_fallback' + } + + +@safe_db_operation +def _clear_sql_model_wrapped(model_class: Type, session=None, **filters) -> DatabaseOperation: + """ + Delete all records from a SQLAlchemy model table, optionally filtered. + Preserves the table structure. + + Args: + model_class: The SQLAlchemy model class to clear + session: Database session (injected by @safe_db_operation decorator) + **filters: Optional column=value filters for selective deletion + e.g., clear_sql_model(User, status='inactive') + + Returns: + DatabaseOperation with deletion count and model name in the .data + + Raises: + AttributeError: If a filter column doesn't exist on the model + """ + query = session.query(model_class) + + # Validate and apply filters + for column_name, value in filters.items(): + if not hasattr(model_class, column_name): + raise AttributeError( + f"Model {model_class.__name__} has no column '{column_name}'" + ) + column = getattr(model_class, column_name) + query = query.filter(column == value) + + # Delete and commit + deleted_count = query.delete(synchronize_session=False) + session.commit() + + return { + 'deleted_rows': deleted_count, + 'model': model_class.__name__, + 'filters_applied': filters if filters else None + } + + + +def clear_sql_model(model_class: Type, **filters) -> dict: + """ + Purpose: + Clear table; fallback to DROP if clear fails + + Role: + Public handler that unwraps DatabaseOperation + """ + result = _clear_sql_model_wrapped(model_class, **filters) + + if result.valid: + return result.data + else: + logger.warning( + f"Failed to clear {model_class.__name__} " + f"({result.error_type}), attempting DROP as fallback..." + ) + + drop_result = _drop_sql_model_wrapped(model_class) + + if drop_result.valid: + return drop_result.data + else: + logger.error( + f"[CLEAR SQL MODEL] Both clear and drop failed for {model_class.__name__}: {drop_result.message}" + ) + return False + \ No newline at end of file diff --git a/core/models/manage/version_check.py b/core/models/manage/version_check.py index 30ca22f..aebd316 100644 --- a/core/models/manage/version_check.py +++ b/core/models/manage/version_check.py @@ -2,7 +2,6 @@ from core.models.orm_models.DatabaseVersion import database_version from core.Constants import Constants from core.errors.logger import logger - # generic from sqlalchemy import exc, text from sqlalchemy.orm import Session @@ -346,4 +345,110 @@ def check_database_compatibility(session: Session) -> dict: "reason": "error", "old_db_version": db_version_you_have, "error": "unknown result" - } \ No newline at end of file + } + + +def delete_sync_metadata(session: Session) -> bool: + """ + Purpose: + Clears existing rows of metadata + + Rank: + Helper + + Called By: + Migrations + + Args: + session: SQLAlchemy session + + Returns: + bool: True if successful, False otherwise. + does NOT raise errors. + """ + func_name = "delete_sync_metadata" # for logging purposes + + # Validate session + if session is None: + logger.error(f"{func_name}: Session is None") + return False + + if not isinstance(session, Session): + logger.error(f"{func_name}: Invalid session type: {type(session)}") + return False + + try: + logger.debug(f"[{func_name}]: Starting transaction for metatable entry delete") + + # Check if table exists before proceeding + try: + session.query(database_version).limit(1).all() + except exc.NoSuchTableError: + logger.info( + f"{func_name}: Table 'database_version' does not exist, this is good because we wanted to clear it." + ) + return True # TRUE! this is great that it's gone already. + + # Delete existing rows + logger.debug(f"{func_name}: Deleting existing version rows..") + deleted_count = session.query(database_version).delete() + logger.debug(f"{func_name}: Deleted {deleted_count} existing row(s)") + + if deleted_count > 1: + logger.warning( + f"{func_name}: Deleted {deleted_count} rows. " + "Table should contain only a single row at any time." + ) + + # Commit transaction + session.commit() + return True + + except exc.IntegrityError as e: + logger.error( + f"{func_name}: Integrity constraint violation (e.g., primary key conflict). " + f"Details: {str(e)}" + ) + session.rollback() + return False + + except exc.OperationalError as e: + logger.info( + f"{func_name}: Sync metadata never existed. " + f"Details: {str(e)}" + ) + session.rollback() + return True # this is good + + except exc.DatabaseError as e: + logger.error( + f"{func_name}: General database error. " + f"Details: {str(e)}" + ) + session.rollback() + return False + + except exc.StatementError as e: + logger.error( + f"{func_name}: SQL statement error. " + f"Details: {str(e)}" + ) + session.rollback() + return False + + except ValueError as e: + logger.error( + f"{func_name}: Value error during metadata deletion." + f"Details: {str(e)}" + ) + session.rollback() + return False + + except Exception as e: + logger.critical( + f"{func_name}: Unexpected exception type {type(e).__name__}. " + f"Details: {str(e)}", + exc_info=True + ) + session.rollback() + return False