begin adding static types to backend code

backend/browser.py

@@ -4,53 +4,70 @@
 # from pprint import pformat
 
 # Partial imports
-from aiohttp import ClientSession, web
-from asyncio import get_event_loop, sleep
-from concurrent.futures import ProcessPoolExecutor
+from aiohttp import ClientSession
+from asyncio import sleep
 from hashlib import sha256
 from io import BytesIO
 from logging import getLogger
-from os import R_OK, W_OK, path, rename, listdir, access, mkdir
+from os import R_OK, W_OK, path, listdir, access, mkdir
 from shutil import rmtree
 from time import time
 from zipfile import ZipFile
 from localplatform import chown, chmod
+from enum import IntEnum
+from typing import Dict, List, TypedDict
 
 # Local modules
-from helpers import get_ssl_context, download_remote_binary_to_path
-from injector import get_gamepadui_tab
+from .loader import Loader, Plugins
+from .helpers import get_ssl_context, download_remote_binary_to_path
+from .settings import SettingsManager
+from .injector import get_gamepadui_tab
 
 logger = getLogger("Browser")
 
+class PluginInstallType(IntEnum):
+    INSTALL = 0
+    REINSTALL = 1
+    UPDATE = 2
+
+class PluginInstallRequest(TypedDict):
+    name: str
+    artifact: str
+    version: str
+    hash: str
+    install_type: PluginInstallType
+
 class PluginInstallContext:
-    def __init__(self, artifact, name, version, hash) -> None:
+    def __init__(self, artifact: str, name: str, version: str, hash: str) -> None:
         self.artifact = artifact
         self.name = name
         self.version = version
         self.hash = hash
 
 class PluginBrowser:
-    def __init__(self, plugin_path, plugins, loader, settings) -> None:
+    def __init__(self, plugin_path: str, plugins: Plugins, loader: Loader, settings: SettingsManager) -> None:
         self.plugin_path = plugin_path
         self.plugins = plugins
         self.loader = loader
         self.settings = settings
-        self.install_requests = {}
+        self.install_requests: Dict[str, PluginInstallContext | List[PluginInstallContext]] = {}
 
-    def _unzip_to_plugin_dir(self, zip, name, hash):
+    def _unzip_to_plugin_dir(self, zip: BytesIO, name: str, hash: str):
         zip_hash = sha256(zip.getbuffer()).hexdigest()
         if hash and (zip_hash != hash):
             return False
         zip_file = ZipFile(zip)
         zip_file.extractall(self.plugin_path)
-        plugin_dir = path.join(self.plugin_path, self.find_plugin_folder(name))
+        plugin_folder = self.find_plugin_folder(name)
+        assert plugin_folder is not None
+        plugin_dir = path.join(self.plugin_path, plugin_folder)
 
         if not chown(plugin_dir) or not chmod(plugin_dir, 555):
             logger.error(f"chown/chmod exited with a non-zero exit code")
             return False
         return True
 
-    async def _download_remote_binaries_for_plugin_with_name(self, pluginBasePath):
+    async def _download_remote_binaries_for_plugin_with_name(self, pluginBasePath: str):
         rv = False
         try:
             packageJsonPath = path.join(pluginBasePath, 'package.json')
@@ -91,7 +108,7 @@ async def _download_remote_binaries_for_plugin_with_name(self, pluginBasePath):
         return rv
 
     """Return the filename (only) for the specified plugin"""
-    def find_plugin_folder(self, name):
+    def find_plugin_folder(self, name: str) -> str | None:
         for folder in listdir(self.plugin_path):
             try:
                 with open(path.join(self.plugin_path, folder, 'plugin.json'), "r", encoding="utf-8") as f:
@@ -102,11 +119,13 @@ def find_plugin_folder(self, name):
             except:
                 logger.debug(f"skipping {folder}")
 
-    async def uninstall_plugin(self, name):
+    async def uninstall_plugin(self, name: str):
         if self.loader.watcher:
             self.loader.watcher.disabled = True
         tab = await get_gamepadui_tab()
-        plugin_dir = path.join(self.plugin_path, self.find_plugin_folder(name))
+        plugin_folder = self.find_plugin_folder(name)
+        assert plugin_folder is not None
+        plugin_dir = path.join(self.plugin_path, )
         try:
             logger.info("uninstalling " + name)
             logger.info(" at dir " + plugin_dir)
@@ -133,7 +152,7 @@ async def uninstall_plugin(self, name):
         if self.loader.watcher:
             self.loader.watcher.disabled = False
 
-    async def _install(self, artifact, name, version, hash):
+    async def _install(self, artifact: str, name: str, version: str, hash: str):
         # Will be set later in code
         res_zip = None
 
@@ -183,6 +202,7 @@ async def _install(self, artifact, name, version, hash):
         ret = self._unzip_to_plugin_dir(res_zip, name, hash)
         if ret:
             plugin_folder = self.find_plugin_folder(name)
+            assert plugin_folder is not None
             plugin_dir = path.join(self.plugin_path, plugin_folder)
             ret = await self._download_remote_binaries_for_plugin_with_name(plugin_dir)
             if ret:
@@ -204,14 +224,14 @@ async def _install(self, artifact, name, version, hash):
         if self.loader.watcher:
             self.loader.watcher.disabled = False
 
-    async def request_plugin_install(self, artifact, name, version, hash, install_type):
+    async def request_plugin_install(self, artifact: str, name: str, version: str, hash: str, install_type: PluginInstallType):
         request_id = str(time())
         self.install_requests[request_id] = PluginInstallContext(artifact, name, version, hash)
         tab = await get_gamepadui_tab()
         await tab.open_websocket()
         await tab.evaluate_js(f"DeckyPluginLoader.addPluginInstallPrompt('{name}', '{version}', '{request_id}', '{hash}', {install_type})")
 
-    async def request_multiple_plugin_installs(self, requests):
+    async def request_multiple_plugin_installs(self, requests: List[PluginInstallRequest]):
         request_id = str(time())
         self.install_requests[request_id] = [PluginInstallContext(req['artifact'], req['name'], req['version'], req['hash']) for req in requests]
         js_requests_parameter = ','.join([
@@ -222,17 +242,17 @@ async def request_multiple_plugin_installs(self, requests):
         await tab.open_websocket()
         await tab.evaluate_js(f"DeckyPluginLoader.addMultiplePluginsInstallPrompt('{request_id}', [{js_requests_parameter}])")
 
-    async def confirm_plugin_install(self, request_id):
+    async def confirm_plugin_install(self, request_id: str):
         requestOrRequests = self.install_requests.pop(request_id)
         if isinstance(requestOrRequests, list):
             [await self._install(req.artifact, req.name, req.version, req.hash) for req in requestOrRequests]
         else:
             await self._install(requestOrRequests.artifact, requestOrRequests.name, requestOrRequests.version, requestOrRequests.hash)
 
-    def cancel_plugin_install(self, request_id):
+    def cancel_plugin_install(self, request_id: str):
         self.install_requests.pop(request_id)
 
-    def cleanup_plugin_settings(self, name):
+    def cleanup_plugin_settings(self, name: str):
         """Removes any settings related to a plugin. Propably called when a plugin is uninstalled.
 
         Args:

backend/helpers.py

@@ -2,13 +2,13 @@
 import ssl
 import uuid
 import os
-import sys
 import subprocess
 from hashlib import sha256
 from io import BytesIO
 
 import certifi
-from aiohttp.web import Response, middleware
+from aiohttp.web import Request, Response, middleware
+from aiohttp.typedefs import Handler
 from aiohttp import ClientSession
 import localplatform
 from customtypes import UserType
@@ -31,17 +31,17 @@ def get_csrf_token():
     return csrf_token
 
 @middleware
-async def csrf_middleware(request, handler):
+async def csrf_middleware(request: Request, handler: Handler):
     if str(request.method) == "OPTIONS" or request.headers.get('Authentication') == csrf_token or str(request.rel_url) == "/auth/token" or str(request.rel_url).startswith("/plugins/load_main/") or str(request.rel_url).startswith("/static/") or str(request.rel_url).startswith("/legacy/") or str(request.rel_url).startswith("/steam_resource/") or str(request.rel_url).startswith("/frontend/") or assets_regex.match(str(request.rel_url)) or frontend_regex.match(str(request.rel_url)):
         return await handler(request)
-    return Response(text='Forbidden', status='403')
+    return Response(text='Forbidden', status=403)
 
 # Get the default homebrew path unless a home_path is specified. home_path argument is deprecated
-def get_homebrew_path(home_path = None) -> str:
+def get_homebrew_path() -> str:
     return localplatform.get_unprivileged_path()
 
 # Recursively create path and chown as user
-def mkdir_as_user(path):
+def mkdir_as_user(path: str):
     path = os.path.realpath(path)
     os.makedirs(path, exist_ok=True)
     localplatform.chown(path)
@@ -57,23 +57,18 @@ def get_loader_version() -> str:
 
 # returns the appropriate system python paths
 def get_system_pythonpaths() -> list[str]:
-    extra_args = {}
-
-    if localplatform.ON_LINUX:
-        # run as normal normal user to also include user python paths
-        extra_args["user"] = localplatform.localplatform._get_user_id()
-        extra_args["env"] = {}
-
     try:
+        # run as normal normal user if on linux to also include user python paths
         proc = subprocess.run(["python3" if localplatform.ON_LINUX else "python", "-c", "import sys; print('\\n'.join(x for x in sys.path if x))"],
-                              capture_output=True, **extra_args)
+        # TODO make this less insane
+                              capture_output=True, user=localplatform.localplatform._get_user_id() if localplatform.ON_LINUX else None, env={} if localplatform.ON_LINUX else None) # type: ignore
         return [x.strip() for x in proc.stdout.decode().strip().split("\n")]
     except Exception as e:
         logger.warn(f"Failed to execute get_system_pythonpaths(): {str(e)}")
         return []
 
 # Download Remote Binaries to local Plugin
-async def download_remote_binary_to_path(url, binHash, path) -> bool:
+async def download_remote_binary_to_path(url: str, binHash: str, path: str) -> bool:
     rv = False
     try:
         if os.access(os.path.dirname(path), os.W_OK):
@@ -110,46 +105,42 @@ def set_user_group() -> str:
 
 # Get the user id hosting the plugin loader
 def get_user_id() -> int:
-    return localplatform.localplatform._get_user_id()
+    return localplatform.localplatform._get_user_id() # pyright: ignore [reportPrivateUsage]
 
 # Get the user hosting the plugin loader
 def get_user() -> str:
-    return localplatform.localplatform._get_user()
+    return localplatform.localplatform._get_user() # pyright: ignore [reportPrivateUsage]
 
 # Get the effective user id of the running process
 def get_effective_user_id() -> int:
-    return localplatform.localplatform._get_effective_user_id()
+    return localplatform.localplatform._get_effective_user_id() # pyright: ignore [reportPrivateUsage]
 
 # Get the effective user of the running process
 def get_effective_user() -> str:
-    return localplatform.localplatform._get_effective_user()
+    return localplatform.localplatform._get_effective_user() # pyright: ignore [reportPrivateUsage]
 
 # Get the effective user group id of the running process
 def get_effective_user_group_id() -> int:
-    return localplatform.localplatform._get_effective_user_group_id()
+    return localplatform.localplatform._get_effective_user_group_id() # pyright: ignore [reportPrivateUsage]
 
 # Get the effective user group of the running process
 def get_effective_user_group() -> str:
-    return localplatform.localplatform._get_effective_user_group()
+    return localplatform.localplatform._get_effective_user_group() # pyright: ignore [reportPrivateUsage]
 
 # Get the user owner of the given file path.
-def get_user_owner(file_path) -> str:
-    return localplatform.localplatform._get_user_owner(file_path)
+def get_user_owner(file_path: str) -> str:
+    return localplatform.localplatform._get_user_owner(file_path) # pyright: ignore [reportPrivateUsage]
 
-# Get the user group of the given file path.
-def get_user_group(file_path) -> str:
-    return localplatform.localplatform._get_user_group(file_path)
+# Get the user group of the given file path, or the user group hosting the plugin loader
+def get_user_group(file_path: str | None = None) -> str:
+    return localplatform.localplatform._get_user_group(file_path) # pyright: ignore [reportPrivateUsage]
 
 # Get the group id of the user hosting the plugin loader
 def get_user_group_id() -> int:
-    return localplatform.localplatform._get_user_group_id()
-
-# Get the group of the user hosting the plugin loader
-def get_user_group() -> str:
-    return localplatform.localplatform._get_user_group()
+    return localplatform.localplatform._get_user_group_id() # pyright: ignore [reportPrivateUsage]
 
 # Get the default home path unless a user is specified
-def get_home_path(username = None) -> str:
+def get_home_path(username: str | None = None) -> str:
     return localplatform.get_home_path(UserType.ROOT if username == "root" else UserType.HOST_USER)
 
 async def is_systemd_unit_active(unit_name: str) -> bool: