Method reset

api_utils/server_state.py:48–97

Reset all state to initial values. Useful for testing.

(self)

        self.reset()

    def reset(self) -> None:
        """Reset all state to initial values. Useful for testing."""
        # --- Stream Queue ---
        self.STREAM_QUEUE: Optional[multiprocessing.Queue] = None
        self.STREAM_PROCESS: Optional[multiprocessing.Process] = None

        # --- Playwright/Browser State ---
        self.playwright_manager: Optional["AsyncPlaywright"] = None
        self.browser_instance: Optional["AsyncBrowser"] = None
        self.page_instance: Optional["AsyncPage"] = None
        self.is_playwright_ready: bool = False
        self.is_browser_connected: bool = False
        self.is_page_ready: bool = False
        self.is_initializing: bool = False

        # --- Proxy Configuration ---
        self.PLAYWRIGHT_PROXY_SETTINGS: Optional[Dict[str, str]] = None

        # --- Model State ---
        self.global_model_list_raw_json: Optional[str] = None
        self.parsed_model_list: List[Dict[str, Any]] = []
        self.model_list_fetch_event: Event = asyncio.Event()
        self.current_ai_studio_model_id: Optional[str] = None
        self.current_auth_profile_path: Optional[str] = None
        self.model_switching_lock: Lock = Lock()
        self.excluded_model_ids: Set[str] = set()

        # --- Request Processing State ---
        self.request_queue: "Optional[Queue[QueueItem]]" = None
        self.processing_lock: Optional[Lock] = None
        self.worker_task: "Optional[Task[None]]" = None

        # --- Parameter Cache ---
        self.page_params_cache: Dict[str, Any] = {}
        self.params_cache_lock: Lock = Lock()

        # --- Debug Logging State ---
        self.console_logs: List[Dict[str, Any]] = []
        self.network_log: Dict[str, List[Dict[str, Any]]] = {
            "requests": [],
            "responses": [],
        }

        # --- Logging ---
        self.logger: logging.Logger = logging.getLogger("AIStudioProxyServer")
        self.log_ws_manager: Optional["WebSocketConnectionManager"] = None

        # --- Control Flags ---
        self.should_exit: bool = False
        self.quota_watchdog: Optional[Callable] = None

    def clear_debug_logs(self) -> None:
        """Clear console and network logs (called after each request)."""