diff --git a/change_log.md b/change_log.md index 88c12a1..57df201 100644 --- a/change_log.md +++ b/change_log.md @@ -1,6 +1,9 @@ # Major Change Log: - +# Sync on No Database Start +### August 5, 2026 +Added a forced sync at startup if there's no DB pre-existing, to handle a bug in GUI with no application versions. Isolated application_version database calls to their own module. And fixed a bug in SyncController to catch if application_version updates and split them into the different tables. +
# Sync Now Integrated & ApplicationVersion Model Transition ### August 4, 2026 diff --git a/core/controllers/ApplicationVersionController.py b/core/controllers/ApplicationVersionController.py index 95d1a4a..c8ea0a7 100644 --- a/core/controllers/ApplicationVersionController.py +++ b/core/controllers/ApplicationVersionController.py @@ -5,7 +5,7 @@ from core.models.session.Application import Application from core.models.orm_models.ApplicationVersion import ApplicationVersion from core.models.manage.wrapper import safe_db_operation, WrapperRollback from core.models.DatabaseOperation import DatabaseOperation, DBErrorType -from core.models.BaseProfile import get_application_version +from core.models.orm_calls.application_version_calls import get_application_version, execute_get_all from core.observers.ApplicationVersionObserver import ApplicationVersionObserver from core.observers.ConnectionObserver import ConnectionObserver @@ -20,16 +20,6 @@ import tarfile from sqlalchemy.orm import Session from sqlalchemy import select -@safe_db_operation -def execute_get_all(application: Optional[Application] = None, session: Session = None) -> DatabaseOperation: - query = select(ApplicationVersion) - - if application is not None: - query = query.where(ApplicationVersion.application_code == application.code) - - result = session.execute(query).scalars().all() - return DatabaseOperation(valid=True, data=result) - class ApplicationVersionController: @staticmethod @@ -45,10 +35,6 @@ class ApplicationVersionController: logger.error(f"[Application Version Controller] Got invalid SQL Query which could not be solved by the wrapper, with error message {database_object.message} and type {database_object.error_type}") return None - # legacy: - # @staticmethod - # def get_all(application: Optional[Application] = None): - # return ApplicationVersion.all(application) @staticmethod def install(application_version: ApplicationVersion, reinstall: bool = False, application_version_observer: Optional[ApplicationVersionObserver] = None, connection_observer: Optional[ConnectionObserver] = None): @@ -149,3 +135,9 @@ class ApplicationVersionController: file.seek(0) return hasher.hexdigest() + + +# legacy: +# @staticmethod +# def get_all(application: Optional[Application] = None): +# return ApplicationVersion.all(application) diff --git a/core/controllers/SyncController.py b/core/controllers/SyncController.py index 7d2e241..56838f8 100644 --- a/core/controllers/SyncController.py +++ b/core/controllers/SyncController.py @@ -90,12 +90,22 @@ def new_sync(client_observer: ClientObserver, connection_observer: ConnectionObs # 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 + observer=connection_observer, + client_observer=client_observer ) if not results.valid: diff --git a/core/models/BaseProfile.py b/core/models/BaseProfile.py index 85afaa8..a1190b3 100644 --- a/core/models/BaseProfile.py +++ b/core/models/BaseProfile.py @@ -22,6 +22,7 @@ from core.models.orm_models.Operator import Operator from core.models.Subscription import Subscription # from core.models.session.ApplicationVersion import ApplicationVersion from core.models.orm_models.ApplicationVersion import ApplicationVersion +from core.models.orm_calls.application_version_calls import get_application_version from dataclasses import dataclass, field, asdict from dataclasses_json import config, Exclude, dataclass_json @@ -49,22 +50,6 @@ def execute_location_sql(country_code: str, city_code: str, session: Session) -> ).scalar_one_or_none() return location_object -@safe_db_operation -def execute_application_sql(application_code: str, version_number: str, session: Session) -> DatabaseOperation: - """Query for a specific application version""" - data = session.query(ApplicationVersion).filter( - ApplicationVersion.application_code == application_code, - ApplicationVersion.version_number == version_number - ).first() - return DatabaseOperation(valid=True, data=data) - - -def get_application_version(application_code: str, version_number: str) -> Optional[ApplicationVersion]: - database_object = execute_application_sql(application_code, version_number) - if database_object.valid: - return database_object.data - else: - return None def get_profile_location_data(country_code: str, city_code: str) -> Location: diff --git a/core/models/DatabaseOperation.py b/core/models/DatabaseOperation.py index ffa6114..f3fa89f 100644 --- a/core/models/DatabaseOperation.py +++ b/core/models/DatabaseOperation.py @@ -31,6 +31,7 @@ class DatabaseOperation: message: Optional[str] = None tried_migration: bool = False tried_filtered: bool = False + which_table: str = None def is_recoverable(self) -> bool: """Can the caller attempt a retry or manual fix?""" diff --git a/core/models/orm_calls/application_version_calls.py b/core/models/orm_calls/application_version_calls.py new file mode 100644 index 0000000..f845c46 --- /dev/null +++ b/core/models/orm_calls/application_version_calls.py @@ -0,0 +1,39 @@ +from core.models.orm_models.ApplicationVersion import ApplicationVersion +from core.models.session.Application import Application +from core.models.manage.wrapper import safe_db_operation, WrapperRollback +from core.models.DatabaseOperation import DatabaseOperation, DBErrorType +from core.errors.logger import logger + +from sqlalchemy.orm import Session +from sqlalchemy import select +from typing import Optional + +@safe_db_operation +def execute_application_sql(application_code: str, version_number: str, session: Session) -> DatabaseOperation: + """Query for a specific application version""" + data = session.query(ApplicationVersion).filter( + ApplicationVersion.application_code == application_code, + ApplicationVersion.version_number == version_number + ).first() + return DatabaseOperation(valid=True, data=data) + + +def get_application_version(application_code: str, version_number: str) -> Optional[ApplicationVersion]: + database_object = execute_application_sql(application_code, version_number) + if database_object.valid: + return database_object.data + else: + logger.error(f"Critical Database error with fetching data: {database_object.error_type}") + return None + + + +@safe_db_operation +def execute_get_all(application: Optional[Application] = None, session: Session = None) -> DatabaseOperation: + query = select(ApplicationVersion) + + if application is not None: + query = query.where(ApplicationVersion.application_code == application.code) + + result = session.execute(query).scalars().all() + return DatabaseOperation(valid=True, data=result) diff --git a/core/services/networking/httpx/async_batch_requests.py b/core/services/networking/httpx/async_batch_requests.py index 5c48c56..5009818 100644 --- a/core/services/networking/httpx/async_batch_requests.py +++ b/core/services/networking/httpx/async_batch_requests.py @@ -1,6 +1,7 @@ from httpx_socks import AsyncProxyTransport from core.services.networking.httpx.async_request import make_async_request from core.services.networking.api_requests.ApiResponseModel import ApiResponse, ErrorType +from core.observers.ClientObserver import ClientObserver from core.errors.logger import logger import json @@ -13,9 +14,11 @@ async def create_async_clearweb_client() -> httpx.AsyncClient: client = httpx.AsyncClient(http2=True, timeout=30) return client -async def create_async_tor_client(port: int, already_checked: bool = False) -> httpx.AsyncClient: +async def create_async_tor_client(client_observer: ClientObserver, port: int, already_checked: bool = False) -> httpx.AsyncClient: - logger.info(f"Creating async client on port {port}") + update_msg = f"Creating async client on port {port}" + logger.info(update_msg) + client_observer.notify('synchronizing', update_msg) """Create a TOR AsyncClient ready for API requests.""" transport = AsyncProxyTransport.from_url(f"socks5://127.0.0.1:{port}") @@ -34,7 +37,9 @@ async def create_async_tor_client(port: int, already_checked: bool = False) -> h data = response.data is_tor = data.get("IsTor", False) if is_tor: - logger.info(f"Tor session initialized successfully on port {port}") + tor_worked_msg = f"Tor session initialized successfully on port {port}" + logger.info(tor_worked_msg) + client_observer.notify('synchronizing', tor_worked_msg) return client else: await client.aclose() @@ -51,20 +56,27 @@ async def create_async_tor_client(port: int, already_checked: bool = False) -> h return None -async def _async_parallel(desired_endpoints: dict, port: int|None = None, already_checked: bool = False) -> dict: +async def _async_parallel(desired_endpoints: dict, client_observer: ClientObserver, port: int|None = None, already_checked: bool = False) -> dict: - # Tor + ############################ + # TOR + ############################ if port is not None: - client = await create_async_tor_client(port, already_checked) + client = await create_async_tor_client(client_observer=client_observer, port=port, already_checked=already_checked) if client is None: return ApiResponse(valid=False, error_type=ErrorType.TOR_NOT_WORKING) - # clearweb: + ############################ + # CLEARWEB + ############################ else: client = await create_async_clearweb_client() + ############################ + # BOTH. RUN BULK ASYNC + ############################ + client_observer.notify('synchronizing', f'Starting Bulk API requests..') start_total = time.time() - try: # Create all coroutines tasks = [ @@ -79,7 +91,9 @@ async def _async_parallel(desired_endpoints: dict, port: int|None = None, alread results = dict(zip(desired_endpoints.keys(), results_list)) total_parallel = time.time() - start_total - logger.info(f"\nTotal time for all API calls: {total_parallel:.2f}s") + total_time_msg = f"\nTotal time for all API calls: {total_parallel:.2f}s" + client_observer.notify('synchronizing', total_time_msg) + logger.info(total_time_msg) return ApiResponse(valid=True, data=results) @@ -87,10 +101,12 @@ async def _async_parallel(desired_endpoints: dict, port: int|None = None, alread await client.aclose() -def async_parallel(desired_endpoints: dict, port: int|None = None, already_checked: bool = False) -> dict: +def async_parallel(desired_endpoints: dict, client_observer: ClientObserver, port: int|None = None, already_checked: bool = False) -> dict: logger.info(f"We got port of {port}") - return asyncio.run(_async_parallel(desired_endpoints, port, already_checked)) - + return asyncio.run(_async_parallel(desired_endpoints, client_observer, port, already_checked)) + + + # has_failures = any(not result.valid for result in results.values()) diff --git a/core/services/networking/httpx/async_request.py b/core/services/networking/httpx/async_request.py index 509e791..fd63b11 100644 --- a/core/services/networking/httpx/async_request.py +++ b/core/services/networking/httpx/async_request.py @@ -1,6 +1,7 @@ from core.services.networking.api_requests.ApiResponseModel import ApiResponse, ErrorType, BackoffStrategy from core.errors.logger import logger from core.services.networking.httpx.classify_response import classify_response +from core.observers.ClientObserver import ClientObserver from typing import Optional import httpx @@ -10,6 +11,9 @@ import time import socket from python_socks._errors import ProxyError + + + async def make_async_request( method: str, url: str, @@ -20,6 +24,8 @@ async def make_async_request( if method == "post" and not payload: return ApiResponse(valid=False, error_type=ErrorType.INVALID_INPUT, message="Can't have a POST request without a payload") + # client_observer.notify('synchronizing', f"Fetching {key}..") + initial_result = await _make_async_request( method=method, url=url, @@ -28,7 +34,10 @@ async def make_async_request( ) if initial_result.valid: + # client_observer.notify('synchronizing', f"Got {key}!") return initial_result + # else: + # client_observer.notify('synchronizing', f"FAILED for {key}! {initial_result.error_type}") logger.error(f"{method.upper()} request failed: {initial_result.error_type}") diff --git a/core/services/networking/httpx/connect.py b/core/services/networking/httpx/connect.py index b9507d9..79571be 100644 --- a/core/services/networking/httpx/connect.py +++ b/core/services/networking/httpx/connect.py @@ -12,6 +12,7 @@ from core.services.networking.api_requests.subtools.extract_domain import extrac from core.controllers.ConfigurationController import ConfigurationController from core.models.Configuration import Configuration, ConnectionChoice +from core.observers.ClientObserver import ClientObserver from core.services.networking.httpx.endpoints import get_endpoints from core.services.networking.httpx.parallel_threading import parallel_thread @@ -183,7 +184,7 @@ def bootstrap_and_try_again(method: str, url: str, observer: ConnectionObserver, -def bulk_async(wanted_list: list, observer: ConnectionObserver) -> ApiResponse: +def bulk_async(wanted_list: list, observer: ConnectionObserver, client_observer: ClientObserver) -> ApiResponse: """ Rank: Orchestrator @@ -207,7 +208,7 @@ def bulk_async(wanted_list: list, observer: ConnectionObserver) -> ApiResponse: ######################################################## if connection_type == ConnectionChoice.SYSTEM: logger.info("We are entering a clearweb async bulk..") - return async_parallel(desired_endpoints=desired_endpoints, port=None) + return async_parallel(desired_endpoints=desired_endpoints, client_observer=client_observer, port=None) ######################################################## @@ -223,12 +224,12 @@ def bulk_async(wanted_list: list, observer: ConnectionObserver) -> ApiResponse: listening = ports.is_port_in_use(_port_used) if listening: - results = async_parallel(desired_endpoints=desired_endpoints, port=_port_used, already_checked=already_checked) + results = async_parallel(desired_endpoints=desired_endpoints, client_observer=client_observer, port=_port_used, already_checked=already_checked) else: new_port = just_only_bootstrap_port(port_used=_port_used, observer=observer) if new_port: _port_used = new_port # Set this as the port for next time - results = async_parallel(desired_endpoints=desired_endpoints, port=new_port, already_checked = False) # Since it's a new port, already_checked is False + results = async_parallel(desired_endpoints=desired_endpoints, client_observer=client_observer, port=new_port, already_checked = False) # Since it's a new port, already_checked is False else: return ApiResponse(valid=False, error_type=ErrorType.TOR_NOT_WORKING) @@ -238,7 +239,7 @@ def bulk_async(wanted_list: list, observer: ConnectionObserver) -> ApiResponse: if results is None: return ApiResponse(valid=False, error_type=ErrorType.UNKNOWN) - return _evaluate_results(results, observer=observer) + return _evaluate_results(results, client_observer=client_observer, observer=observer) def just_only_bootstrap_port(port_tried: int, observer: ConnectionObserver) -> ApiResponse|int: @@ -250,7 +251,7 @@ def just_only_bootstrap_port(port_tried: int, observer: ConnectionObserver) -> A return False -def _evaluate_results(results: ApiResponse, observer: ConnectionObserver) -> ApiResponse: +def _evaluate_results(results: ApiResponse, client_observer: ClientObserver, observer: ConnectionObserver) -> ApiResponse: """ Purpose: ALL async bulk results flow through here. Good or bad. @@ -311,7 +312,7 @@ def _evaluate_results(results: ApiResponse, observer: ConnectionObserver) -> Api tor_endpoints = get_endpoints(wanted_list=tor_problem_list, dns_resolver=False) new_port = just_only_bootstrap_port(port_tried=_port_used, observer=observer) if new_port: - return async_parallel(desired_endpoints=tor_endpoints, port=new_port, already_checked = False) # new port means already_checked = False + return async_parallel(desired_endpoints=tor_endpoints, client_observer=client_observer, port=new_port, already_checked = False) # new port means already_checked = False else: # Preserve working results—caller sees what *did* work return ApiResponse(valid=False, data=working_results, error_type=ErrorType.TOR_NOT_WORKING) diff --git a/core/services/sync/legacy_insert.py b/core/services/sync/legacy_insert.py index 4fd673b..afc2580 100644 --- a/core/services/sync/legacy_insert.py +++ b/core/services/sync/legacy_insert.py @@ -64,24 +64,24 @@ def for_subscriptions(api_response: dict) -> DatabaseOperation: # "librewolf": 5 # } -def for_application_versions(api_response: dict, code: str) -> DatabaseOperation: - try: - application_versions = [] - for application_version in api_response['data']: - # print(f"application_version['format_revision'] is {application_version['format_revision']}") - app = ApplicationVersion(code, - application_version['version_number'], - application_version['format_revision'], - application_version['id'], - application_version['download_path'], - application_version['released_at'], - application_version['file_hash']) - application_versions.append(app) +# def for_application_versions(api_response: dict, code: str) -> DatabaseOperation: +# try: +# application_versions = [] +# for application_version in api_response['data']: +# # print(f"application_version['format_revision'] is {application_version['format_revision']}") +# app = ApplicationVersion(code, +# application_version['version_number'], +# application_version['format_revision'], +# application_version['id'], +# application_version['download_path'], +# application_version['released_at'], +# application_version['file_hash']) +# application_versions.append(app) - ApplicationVersion.truncate() - ApplicationVersion.save_many(application_versions) +# ApplicationVersion.truncate() +# ApplicationVersion.save_many(application_versions) - return DatabaseOperation(valid=True) - except Exception as e: - logger.info(f"Error with legacy for_application_versions insert: {str(e)}") - return DatabaseOperation(valid=False, error_type= DBErrorType.UNKNOWN, message=str(e)) +# return DatabaseOperation(valid=True) +# except Exception as e: +# logger.info(f"Error with legacy for_application_versions insert: {str(e)}") +# return DatabaseOperation(valid=False, error_type= DBErrorType.UNKNOWN, message=str(e))