From 81002cd76d2c048c4160834e285c48ac75aee3ba Mon Sep 17 00:00:00 2001 From: SimplifiedPrivacy Date: Wed, 14 Oct 2026 18:16:13 -0400 Subject: [PATCH] The Plumbing now allows for the service to provide it's own function_choice when it returns a result. Separetly, a new type_checker wrapper is now available for use. And finally, the config model now passes around the same instance as a singleton to avoid new file operations to just read --- core/controllers/ProfileController.py | 4 +- core/infrastructure/plumbing.py | 5 +- core/models/Configuration.py | 71 ++++++++++++++++----------- core/models/Result.py | 3 ++ core/utils/confirm_preqs.py | 14 ++++++ core/utils/type_checker.py | 43 ++++++++++++++++ 6 files changed, 110 insertions(+), 30 deletions(-) create mode 100644 core/utils/confirm_preqs.py create mode 100644 core/utils/type_checker.py diff --git a/core/controllers/ProfileController.py b/core/controllers/ProfileController.py index b6f6321..8e36f67 100644 --- a/core/controllers/ProfileController.py +++ b/core/controllers/ProfileController.py @@ -61,7 +61,7 @@ class ProfileController: connection_observer: ConnectionObserver = None, ticket_observer: TicketObserver = None, max_resolution: Optional[str] = None - ): + ) -> Result: from core.controllers.ConnectionController import ConnectionController @@ -113,6 +113,7 @@ class ProfileController: profile_observer.notify('enabled', profile) ApplicationController.launch(application_version, profile, port_number, asynchronous=asynchronous, profile_observer=profile_observer, ticket_observer=ticket_observer, connection_observer=connection_observer) + return Result(valid=True) # ============================================================================ # SYSTEMWIDE @@ -123,6 +124,7 @@ class ProfileController: if connection_result.valid: if profile_observer is not None: profile_observer.notify('enabled', profile) + return Result(valid=True) else: logger.error(f"Couldn't enable the profile: {connection_result.error_type}") error_msg = connection_result.message diff --git a/core/infrastructure/plumbing.py b/core/infrastructure/plumbing.py index 74c9e78..f046a93 100644 --- a/core/infrastructure/plumbing.py +++ b/core/infrastructure/plumbing.py @@ -56,7 +56,10 @@ class ThreadPlumbing: if not isinstance(result, Result): result = convert_to_result(result) if not result.valid: - result.function_choice = resolve_prescription(result, self.ENUM_MAP, *self.args, **self.kwargs) + if not result.function_choice: + result.function_choice = resolve_prescription(result, self.ENUM_MAP, *self.args, **self.kwargs) + else: + logger.info("Using the pre-existing function choice for this error.") except Exception as e: error_msg = f"There was an error: {str(e)}" # notify(error_msg) diff --git a/core/models/Configuration.py b/core/models/Configuration.py index 412a407..a383c94 100644 --- a/core/models/Configuration.py +++ b/core/models/Configuration.py @@ -8,17 +8,23 @@ from enum import Enum from pydantic import BaseModel, field_serializer, field_validator, ConfigDict from datetime import datetime from zoneinfo import ZoneInfo -from typing import Optional, Self +from typing import Optional, Self, ClassVar import json import os import sys - class ConnectionChoice(str, Enum): TOR = "tor" SYSTEM = "system" + class Configuration(BaseModel): + model_config = ConfigDict( + extra='ignore', + exclude_none=True + ) + + _instance: ClassVar[Optional[Self]] = None connection: Optional[ConnectionChoice] = None auto_sync_enabled: Optional[bool] = None endpoint_verification_enabled: Optional[bool] = False @@ -28,11 +34,8 @@ class Configuration(BaseModel): did_sudo_setup: Optional[bool] = False singbox: Optional[str] = None - model_config = ConfigDict( - extra='ignore', # Ignore unknown fields in JSON - exclude_none=True # Don't serialize None values - ) - + + @field_validator('last_synced_at', mode='before') @classmethod def parse_datetime(cls, v): @@ -47,32 +50,58 @@ class Configuration(BaseModel): return value.isoformat().replace('+00:00', 'Z') # +00:00 → Z for JSON return None + @classmethod + def get(cls) -> Optional[Self]: + # Return the cached instance if it exists + if cls._instance is not None: + return cls._instance - @staticmethod - def get(): try: with open(f'{Constants.HV_CONFIG_HOME}/config.json', 'r') as f: config_file_contents = f.read() except FileNotFoundError: return None - + try: configuration_dict = json.loads(config_file_contents) except ValueError: sys.exit(1) - - return Configuration(**configuration_dict) # Pydantic validates on init + # Create the instance and cache it before returning + cls._instance = cls(**configuration_dict) + return cls._instance - def save(self: Self): + def save(self) -> None: + # model_dump_json reflects the current state of this specific instance config_file_contents = f'{self.model_dump_json(indent=4)}\n' - os.makedirs(Constants.HV_CONFIG_HOME, exist_ok=True) + os.makedirs(Constants.HOME, exist_ok=True) config_file_path = f'{Constants.HV_CONFIG_HOME}/config.json' write_atomically(config_file_path, config_file_contents) +def get_setting(looking_for: str): + # Use the singleton getter + config = Configuration.get() + + if not config: + logger.error(f"[CONFIG] Can't load the entire config") + return None + + # getattr because Pydantic models are objects, not dicts + result = getattr(config, looking_for, None) + + if result is None: + logger.error(f"[CONFIG] What you want isn't in the config: {looking_for}") + return None + + return result + + def read_config(): + """ + Returns it as a dict, not as a model. + """ try: config_file_contents = open(f'{Constants.HV_CONFIG_HOME}/config.json', 'r').read() except FileNotFoundError: @@ -84,17 +113,3 @@ def read_config(): logger.error(f"[CONFIG] Can't load JSON config") return configuration - -def get_setting(looking_for): - config = read_config() - if not config: - logger.error(f"[CONFIG] Can't load the entire config") - return None - - if looking_for not in config: - logger.error(f"[CONFIG] What you want isn't in the config") - return None - - result = config[looking_for] - - return result diff --git a/core/models/Result.py b/core/models/Result.py index e4965a0..fe28c4a 100644 --- a/core/models/Result.py +++ b/core/models/Result.py @@ -35,6 +35,7 @@ class ResultError(Enum): TIMEOUT = "timeout" INVALID_API_REPLY = "invalid_api_reply" ENCRYPTION = "encryption" + ENDPOINT_VERIFICATION = "endpoint_verification" LEAK_ISSUE = "leak_issue" NEED_SYNC = "need_sync" WRONG_LOCATION = "wrong_location" @@ -42,6 +43,8 @@ class ResultError(Enum): UNPAID = "unpaid" MTU = "mtu" ACTIVE_BILLING = "active_billing" + STATE_CONFLICT = "state_conflict" + TOR_NEEDED = "tor_needed" NAVIGATE_TO = "navigate_to" UNKNOWN = "unknown" diff --git a/core/utils/confirm_preqs.py b/core/utils/confirm_preqs.py new file mode 100644 index 0000000..71193e4 --- /dev/null +++ b/core/utils/confirm_preqs.py @@ -0,0 +1,14 @@ +from core.models.Result import Result, ResultError + +from typing import Optional + +def confirm_prereqs(PRE_REQS: list, allow_false: Optional[bool] = False) -> Result: + for each_prereq in PRE_REQS: + if allow_false: + if each_prereq == None: + return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=f"You are missing {each_prereq}") + else: + if not each_prereq: + return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=f"You are missing {each_prereq}") + return Result(valid=True) + diff --git a/core/utils/type_checker.py b/core/utils/type_checker.py new file mode 100644 index 0000000..53e85a3 --- /dev/null +++ b/core/utils/type_checker.py @@ -0,0 +1,43 @@ +from core.models.Result import Result + +import inspect +from functools import wraps +from typing import Any + +def type_checker(func): + """ + Type check wrapper. + This type checks individual functions it's put on, and returns a Result object on failure. + """ + @wraps(func) + def wrapper(*args, **kwargs): + # 1. Introspect the wrapped function + sig = inspect.signature(func) + + try: + # Bind the provided args/kwargs to the function's signature + bound_args = sig.bind(*args, **kwargs) + bound_args.apply_defaults() + except TypeError as e: + # This handles cases where the wrong number of arguments are passed + return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=str(e)) + + # 2. Perform type validation + for name, value in bound_args.arguments.items(): + expected_type = sig.parameters[name].annotation + + # Skip parameters that don't have a type hint + if expected_type is inspect.Parameter.empty: + continue + + if not isinstance(value, expected_type): + return Result( + valid=False, + error_type=ResultError.INVALID_INPUT, + message=f"Argument '{name}' must be {expected_type.__name__}, got type {type(value).__name__}" + ) + + # 3. If all checks pass, call the original function + return func(*args, **kwargs) + + return wrapper