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.insert_for_orm import insert_one_orm_model from core.services.networking.httpx import connect from core.services.networking.api_requests.ApiResponseModel import ApiResponse, ErrorType from core.models.DatabaseOperation import DatabaseOperation, DBErrorType from core.models.Result import Result, ResultError from core.errors.logger import logger # ORM models that can be sync'ed: from core.models.orm_models.Location import Location from core.models.orm_models.Operator import Operator from core.Constants import Constants from core.services.sync import legacy_insert from core.controllers.ApplicationController import ApplicationController from core.controllers.ApplicationVersionController import ApplicationVersionController from core.controllers.ClientVersionController import ClientVersionController from core.controllers.ConfigurationController import ConfigurationController from core.controllers.SubscriptionPlanController import SubscriptionPlanController from core.observers.ClientObserver import ClientObserver from core.observers.ConnectionObserver import ConnectionObserver from core.models.orm_models.ApplicationVersion import ApplicationVersion import sys import json ORM_TABLES = { "locations": Location, "operators": Operator, "application_versions": ApplicationVersion } LEGACY_SQL_FUNCT_DICT = { "applications": legacy_insert.for_applications, "client_version": legacy_insert.for_client_version, "subscriptions": legacy_insert.for_subscriptions, } APP_CODES = { "firefox": 1, "chromium": 2, "brave": 3, "librewolf": 5 } def call_legacy_insert_function(key: str, new_data: dict): func = LEGACY_SQL_FUNCT_DICT.get(key) if func is None: return DatabaseOperation(valid=False, error_type=DBErrorType.UNKNOWN_MODEL) if key in APP_CODES: code = APP_CODES[key] return func(new_data, code) else: return func(new_data) def get_orm_model(key) -> bool: if key not in ORM_TABLES: return False return ORM_TABLES[key] def new_sync(client_observer: ClientObserver, connection_observer: ConnectionObserver) -> Result: client_observer.notify('synchronizing', "Fetching list of new data ..") #################################### # METADATA. Should we even sync? #################################### metadata_result = coordinate_cache_sync(client_observer, connection_observer) # Outright Error: if not metadata_result["success"]: error_msg = metadata_result["error"] client_observer.notify('synchronizing', f'Error! {error_msg}') return Result(valid=False) # Nothing changed if the 'changed_tables' variable does NOT exist changed_tables = metadata_result["changed_tables"] if not changed_tables: client_observer.notify('synchronized') return Result(valid=True) # We only make it past this point if there's New Data #################################### # MAKE SURE TO GET NEW APPS (subgroups) #################################### if 'application_versions' in changed_tables: logger.info("Adding the application_versions to changed tables!") changed_tables.extend(APP_CODES.keys()) print(f"changed_tables is {changed_tables}") #################################### # API CALLS: GET NEW DATA IN BULK #################################### results = connect.bulk_async( wanted_list=changed_tables, observer=connection_observer, client_observer=client_observer ) if not results.valid: error_msg = f"Sync failed! {results.error_type}" logger.error(f"{error_msg} {results.message}") client_observer.notify('synchronizing', f'Error! {error_msg}') return Result(valid=False) quantity_of_entries = len(results.data) logger.info(f"We have valid API call results. There are {quantity_of_entries} entries") #################################### # LOOP INSERT INTO DATABASE #################################### client_observer.notify('synchronizing', f'Inserting into Database..') skipped = [] FIRST_APP_VERSION_LOOP_ITERATION = True for key, each_api_result in results.data.items(): # logger.info(f"We are inserting {key}") if not each_api_result.valid: logger.info(f"Skipping invalid api response for {key}") skipped.append(key) continue logger.info(f"Inserting valid api data for {key} into the Database, and FIRST_APP_VERSION_LOOP_ITERATION is {FIRST_APP_VERSION_LOOP_ITERATION}") each_insert = insert_data( key=key, each_api_calls_data=each_api_result.data, client_observer=client_observer, FIRST_APP_VERSION_LOOP_ITERATION=FIRST_APP_VERSION_LOOP_ITERATION, ) logger.info(f"Exited the insert data function with a result of {each_insert.valid}") if each_insert.which_table == "application_versions": FIRST_APP_VERSION_LOOP_ITERATION = False if each_insert.valid: logger.info(f"Success with insert of {key}") client_observer.notify('synchronizing', f'Inserted {key}') continue else: skipped.append(key) logger.error(f"Database error with inserting {key}, because {each_insert.error_type}") client_observer.notify('synchronizing', f'Failed inserting {key} because {each_insert.message}') logger.info("This only gets triggered in errors, but Moving on to the next item..") #################################### # FINAL EVALUATION #################################### total_skipped = len(skipped) logger.info(f"We exited the loop, doing the final evaluation. And a total of {total_skipped} were skipped.") ConfigurationController.update_last_synced_at() filtered_metadata = metadata_result["filtered_metadata"] # from the top of the function if total_skipped == 0: client_observer.notify('synchronized', "Fetch & Save Complete!") save_successful = save_metadata(filtered_metadata) if save_successful: return Result(valid=True, message="Finshed sync.") else: error_msg = "Finshed sync, but had issues with the saving of metadata for next time." logger.error(error_msg) return Result(valid=True, message=error_msg) elif total_skipped < quantity_of_entries: error_msg = f"Partial Success. {total_skipped} skipped." client_observer.notify('synchronized', error_msg) return Result(valid=True, data=skipped, message=error_msg) else: error_msg = f"Sync Failed. All {total_skipped} entries were skipped!" client_observer.notify('synchronized', error_msg) return Result(valid=False, data=skipped, message="Complete Failure, all data failed to insert.") def insert_data( key: str, each_api_calls_data: dict, client_observer: ClientObserver, FIRST_APP_VERSION_LOOP_ITERATION: bool ) -> DatabaseOperation: #################################### # NEW ORM SYSTEM #################################### new_orm_model = get_orm_model(key) if new_orm_model: db_result = insert_one_orm_model( which_key=key, which_model=new_orm_model, new_data=each_api_calls_data, override=True ) return db_result # application versions if key in APP_CODES: logger.info(f"{key} counts as a new ORM but uses 'application_versions' and FIRST_APP_VERSION_LOOP_ITERATION is {FIRST_APP_VERSION_LOOP_ITERATION}") new_orm_model = get_orm_model("application_versions") db_result = insert_one_orm_model( which_key="application_versions", which_model=new_orm_model, new_data=each_api_calls_data, override=FIRST_APP_VERSION_LOOP_ITERATION, ) db_result.which_table = "application_versions" return db_result #################################### # LEGACY MANUAL SQL #################################### return call_legacy_insert_function(key=key, new_data=each_api_calls_data) #################################### # UNKNOWN MODEL #################################### return DatabaseOperation(valid=False, error_type=DBErrorType.UNKNOWN_MODEL) # Legacy version: # "firefox": legacy_insert.for_application_versions, # "chromium": legacy_insert.for_application_versions, # "brave": legacy_insert.for_application_versions, # "librewolf": legacy_insert.for_application_versions, # ISOLATION TESTS: # changed_tables = ["locations", "operators", "applications", "client_version", "subscriptions", "applications", "firefox", "chromium", "brave", "librewolf"]