diff --git a/core/controllers/ClientController.py b/core/controllers/ClientController.py index 759fc4a..a0128e7 100644 --- a/core/controllers/ClientController.py +++ b/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: diff --git a/core/models/DatabaseOperation.py b/core/models/DatabaseOperation.py new file mode 100644 index 0000000..65bb033 --- /dev/null +++ b/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") diff --git a/core/models/manage/insert.py b/core/models/manage/insert.py index 29f20b4..84b81eb 100644 --- a/core/models/manage/insert.py +++ b/core/models/manage/insert.py @@ -1,20 +1,28 @@ -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: """ - Generic ORM insert for any model. Keeping it generic for reuse. + Purpose: + Generic ORM insert for any model. Keeping it generic for reuse. Args: model_class: The ORM model class @@ -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() + # Normalize to list for uniform handling + data_list = all_data if isinstance(all_data, list) else [all_data] - logger.info(f"Starting insert for {model_class.__name__}") + # Call the wrapped function with normalized data + return _wrapped_insert(model_class=model_class, data_list=data_list, override=override) - # Normalize to list for uniform handling - data_list = all_data if isinstance(all_data, list) else [all_data] +@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 + return DatabaseOperation(valid=True) + + except TypeError as 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() + + valid_fields = {col.name for col in model_class.__table__.columns} + filtered_data_list = [ + {k: v for k, v in each_json.items() if k in valid_fields} + for each_json in data_list + ] + + dropped_fields = set().union(*(set(d.keys()) - valid_fields for d in data_list)) + if dropped_fields: + logger.warning(f"Old Client Dropped unknown fields: {dropped_fields}") + try: - for each_json in data_list: - instance = model_class(**each_json) + for filtered_json in filtered_data_list: + instance = model_class(**filtered_json) session.add(instance) - session.commit() - logger.info(f"Completed! SQL Insertion Successful following the {model_class.__name__} Model’s Type Rules") - return True + + # 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}") + )) - except TypeError as e: - print(f"TypeError caught: {e}") - session.rollback() - raise ValueError(f"Invalid fields for {model_class.__name__}: {e}") - except IntegrityError as e: - print(f"IntegrityError caught: {e.orig}") - session.rollback() - raise ValueError(f"Constraint violation: {e.orig}") - except SQLAlchemyError as e: - print(f"SQLAlchemyError caught: {e}") - session.rollback() - raise - except Exception as e: - print(f"Generic exception caught: {e}") - session.rollback() - raise e + +# =================== 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 diff --git a/core/models/manage/migrations.py b/core/models/manage/migrations.py new file mode 100644 index 0000000..88417e7 --- /dev/null +++ b/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)}" + ) diff --git a/core/models/manage/session_management.py b/core/models/manage/session_management.py index 8170644..f0c111a 100644 --- a/core/models/manage/session_management.py +++ b/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.") + 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}") - # if using without checking or try except blocks: - # database_version.create(engine, checkfirst=True) + +# 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(): diff --git a/core/models/manage/wrapper.py b/core/models/manage/wrapper.py new file mode 100644 index 0000000..8ce96e4 --- /dev/null +++ b/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 + diff --git a/core/models/orm_models/Location.py b/core/models/orm_models/Location.py index 85621fb..68ce7a5 100644 --- a/core/models/orm_models/Location.py +++ b/core/models/orm_models/Location.py @@ -45,7 +45,7 @@ class Location(BaseModel): is_wireguard_capable: Mapped[Optional[bool]] = mapped_column(String, nullable=True, default=None) is_hysteria2_capable: Mapped[Optional[bool]] = mapped_column(String, nullable=True, default=None) is_vless_capable: Mapped[Optional[bool]] = mapped_column(String, nullable=True, default=None) - + def to_dict(self): return { "country_code": self.country_code, diff --git a/core/services/sync/orm_methods/sync_one_orm.py b/core/services/sync/orm_methods/sync_one_orm.py index 8af5912..ff55b59 100644 --- a/core/services/sync/orm_methods/sync_one_orm.py +++ b/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: - did_it_work = insert_into_model(which_model, denormalized_data, True) - if did_it_work: - 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}" - } - except: + database_operation_object = insert_into_model(which_model, denormalized_data, True) + + if database_operation_object.valid: + return { + "success": True + } + else: 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}" + "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}" } - else: return { "success": False, + "category": "Data", "error": f"We had issues with denormalizing the data from the server. That data was {new_data}." }