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.
This commit is contained in:
parent
46c8721a1a
commit
b70818f774
10 changed files with 127 additions and 71 deletions
|
|
@ -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.
|
||||
</br>
|
||||
|
||||
# Sync Now Integrated & ApplicationVersion Model Transition
|
||||
### August 4, 2026
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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?"""
|
||||
|
|
|
|||
39
core/models/orm_calls/application_version_calls.py
Normal file
39
core/models/orm_calls/application_version_calls.py
Normal file
|
|
@ -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)
|
||||
|
|
@ -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())
|
||||
|
|
|
|||
|
|
@ -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}")
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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))
|
||||
|
|
|
|||
Loading…
Reference in a new issue