From eeb18eb0176a19f305972f6b2c985aa26e67d2a5 Mon Sep 17 00:00:00 2001 From: kramo Date: Wed, 16 Aug 2023 19:18:03 +0200 Subject: [PATCH] Typing --- src/details_window.py | 4 +- src/errors/error_producer.py | 6 +-- src/errors/friendly_error.py | 6 +-- src/game.py | 8 ++-- src/game_cover.py | 20 ++++----- src/importer/importer.py | 67 ++++++++++++++++------------- src/importer/sources/location.py | 10 +++-- src/importer/sources/source.py | 12 +++--- src/logging/color_log_formatter.py | 2 +- src/logging/session_file_handler.py | 10 +++-- src/logging/setup.py | 4 +- src/main.py | 6 +-- src/preferences.py | 4 +- src/store/managers/manager.py | 12 +++--- src/store/pipeline.py | 2 +- src/store/store.py | 12 +++--- src/window.py | 6 +-- 17 files changed, 102 insertions(+), 89 deletions(-) diff --git a/src/details_window.py b/src/details_window.py index ead7710..bb234da 100644 --- a/src/details_window.py +++ b/src/details_window.py @@ -19,7 +19,7 @@ import os from time import time -from typing import Any +from typing import Any, Optional from gi.repository import Adw, Gio, GLib, Gtk from PIL import Image @@ -55,7 +55,7 @@ class DetailsWindow(Adw.Window): cover_changed: bool = False - def __init__(self, game: Game | None = None, **kwargs: Any): + def __init__(self, game: Optional[Game] = None, **kwargs: Any): super().__init__(**kwargs) self.game: Game = game diff --git a/src/errors/error_producer.py b/src/errors/error_producer.py index 7f31fea..f3c9b53 100644 --- a/src/errors/error_producer.py +++ b/src/errors/error_producer.py @@ -8,14 +8,14 @@ class ErrorProducer: Specifies the report_error and collect_errors methods in a thread-safe manner. """ - errors: list[Exception] = None - errors_lock: Lock = None + errors: list[Exception] + errors_lock: Lock def __init__(self) -> None: self.errors = [] self.errors_lock = Lock() - def report_error(self, error: Exception): + def report_error(self, error: Exception) -> None: """Report an error""" with self.errors_lock: self.errors.append(error) diff --git a/src/errors/friendly_error.py b/src/errors/friendly_error.py index 4c9ca7f..586d784 100644 --- a/src/errors/friendly_error.py +++ b/src/errors/friendly_error.py @@ -1,4 +1,4 @@ -from typing import Iterable +from typing import Iterable, Optional class FriendlyError(Exception): @@ -27,8 +27,8 @@ class FriendlyError(Exception): self, title: str, subtitle: str, - title_args: Iterable[str] = None, - subtitle_args: Iterable[str] = None, + title_args: Optional[Iterable[str]] = None, + subtitle_args: Optional[Iterable[str]] = None, ) -> None: """Create a friendly error diff --git a/src/game.py b/src/game.py index edeaad4..e144460 100644 --- a/src/game.py +++ b/src/game.py @@ -23,7 +23,7 @@ import shlex import subprocess from pathlib import Path from time import time -from typing import Any +from typing import Any, Optional from gi.repository import Adw, GLib, GObject, Gtk @@ -57,7 +57,7 @@ class Game(Gtk.Box): hidden: bool = False last_played: int = 0 name: str - developer: str | None = None + developer: Optional[str] = None removed: bool = False blacklisted: bool = False game_cover: GameCover = None @@ -97,7 +97,7 @@ class Game(Gtk.Box): def save(self) -> None: self.emit("save-ready", {}) - def create_toast(self, title: str, action: str | None = None) -> None: + def create_toast(self, title: str, action: Optional[str] = None) -> None: toast = Adw.Toast.new(title.format(self.name)) toast.set_priority(Adw.ToastPriority.HIGH) @@ -180,7 +180,7 @@ class Game(Gtk.Box): self.cover.set_opacity(int(not loading)) self.spinner.set_spinning(loading) - def get_cover_path(self) -> Path | None: + def get_cover_path(self) -> Optional[Path]: cover_path = shared.covers_dir / f"{self.game_id}.gif" if cover_path.is_file(): return cover_path # type: ignore diff --git a/src/game_cover.py b/src/game_cover.py index 24ebf09..221075d 100644 --- a/src/game_cover.py +++ b/src/game_cover.py @@ -18,7 +18,7 @@ # SPDX-License-Identifier: GPL-3.0-or-later from pathlib import Path -from typing import Any, Callable +from typing import Any, Callable, Optional from gi.repository import Gdk, GdkPixbuf, Gio, GLib, Gtk from PIL import Image, ImageFilter, ImageStat @@ -27,12 +27,12 @@ from src import shared class GameCover: - texture: Gdk.Texture | None - blurred: Gdk.Texture | None - luminance: tuple[float, float] | None - path: Path | None - animation: GdkPixbuf.PixbufAnimation | None - anim_iter: GdkPixbuf.PixbufAnimationIter | None + texture: Optional[Gdk.Texture] + blurred: Optional[Gdk.Texture] + luminance: Optional[tuple[float, float]] + path: Optional[Path] + animation: Optional[GdkPixbuf.PixbufAnimation] + anim_iter: Optional[GdkPixbuf.PixbufAnimationIter] placeholder = Gdk.Texture.new_from_resource( shared.PREFIX + "/library_placeholder.svg" @@ -41,12 +41,12 @@ class GameCover: shared.PREFIX + "/library_placeholder_small.svg" ) - def __init__(self, pictures: set[Gtk.Picture], path: Path | None = None) -> None: + def __init__(self, pictures: set[Gtk.Picture], path: Optional[Path] = None) -> None: self.pictures = pictures self.new_cover(path) # Wrap the function in another one as Gio.Task.run_in_thread does not allow for passing args - def create_func(self, path: Path | None) -> Callable: + def create_func(self, path: Optional[Path]) -> Callable: self.animation = GdkPixbuf.PixbufAnimation.new_from_file(str(path)) self.anim_iter = self.animation.get_iter() @@ -55,7 +55,7 @@ class GameCover: return wrapper - def new_cover(self, path: Path | None = None) -> None: + def new_cover(self, path: Optional[Path] = None) -> None: self.animation = None self.texture = None self.blurred = None diff --git a/src/importer/importer.py b/src/importer/importer.py index dbcc715..e987760 100644 --- a/src/importer/importer.py +++ b/src/importer/importer.py @@ -19,6 +19,7 @@ # SPDX-License-Identifier: GPL-3.0-or-later import logging +from typing import Any, Optional from gi.repository import Adw, GLib, Gtk @@ -37,22 +38,22 @@ from src.utils.task import Task class Importer(ErrorProducer): """A class in charge of scanning sources for games""" - progressbar = None - import_statuspage = None - import_dialog = None - summary_toast = None + progressbar: Gtk.ProgressBar + import_statuspage: Adw.StatusPage + import_dialog: Adw.MessageDialog + summary_toast: Adw.Toast - sources: set[Source] = None + sources: set[Source] n_source_tasks_created: int = 0 n_source_tasks_done: int = 0 n_pipelines_done: int = 0 - game_pipelines: set[Pipeline] = None + game_pipelines: set[Pipeline] removed_game_ids: set[str] = set() imported_game_ids: set[str] = set() - def __init__(self): + def __init__(self) -> None: super().__init__() # TODO: make this stateful @@ -63,23 +64,23 @@ class Importer(ErrorProducer): self.sources = set() @property - def n_games_added(self): + def n_games_added(self) -> int: return sum( 1 if not (pipeline.game.blacklisted or pipeline.game.removed) else 0 for pipeline in self.game_pipelines ) @property - def pipelines_progress(self): + def pipelines_progress(self) -> float: progress = sum(pipeline.progress for pipeline in self.game_pipelines) try: progress = progress / len(self.game_pipelines) except ZeroDivisionError: progress = 0 - return progress + return progress # type: ignore @property - def sources_progress(self): + def sources_progress(self) -> float: try: progress = self.n_source_tasks_done / self.n_source_tasks_created except ZeroDivisionError: @@ -87,16 +88,16 @@ class Importer(ErrorProducer): return progress @property - def finished(self): + def finished(self) -> bool: return ( self.n_source_tasks_created == self.n_source_tasks_done and len(self.game_pipelines) == self.n_pipelines_done ) - def add_source(self, source): + def add_source(self, source: Source) -> None: self.sources.add(source) - def run(self): + def run(self) -> None: """Use several Gio.Task to import games from added sources""" shared.win.get_application().lookup_action("import").set_enabled(False) @@ -121,7 +122,7 @@ class Importer(ErrorProducer): self.progress_changed_callback() - def create_dialog(self): + def create_dialog(self) -> None: """Create the import dialog""" self.progressbar = Gtk.ProgressBar(margin_start=12, margin_end=12) self.import_statuspage = Adw.StatusPage( @@ -138,7 +139,9 @@ class Importer(ErrorProducer): ) self.import_dialog.present() - def source_task_thread_func(self, _task, _obj, data, _cancellable): + def source_task_thread_func( + self, _task: Any, _obj: Any, data: tuple, _cancellable: Any + ) -> None: """Source import task code""" source: Source @@ -192,27 +195,27 @@ class Importer(ErrorProducer): pipeline.connect("advanced", self.pipeline_advanced_callback) self.game_pipelines.add(pipeline) - def update_progressbar(self): + def update_progressbar(self) -> None: """Update the progressbar to show the overall import progress""" # Reserve 10% for the sources discovery, the rest is the pipelines self.progressbar.set_fraction( (0.1 * self.sources_progress) + (0.9 * self.pipelines_progress) ) - def source_callback(self, _obj, _result, data): + def source_callback(self, _obj: Any, _result: Any, data: tuple) -> None: """Callback executed when a source is fully scanned""" source, *_rest = data logging.debug("Import done for source %s", source.source_id) self.n_source_tasks_done += 1 self.progress_changed_callback() - def pipeline_advanced_callback(self, pipeline: Pipeline): + def pipeline_advanced_callback(self, pipeline: Pipeline) -> None: """Callback called when a pipeline for a game has advanced""" if pipeline.is_done: self.n_pipelines_done += 1 self.progress_changed_callback() - def progress_changed_callback(self): + def progress_changed_callback(self) -> None: """ Callback called when the import process has progressed @@ -225,7 +228,7 @@ class Importer(ErrorProducer): if self.finished: self.import_callback() - def remove_games(self): + def remove_games(self) -> None: """Set removed to True for missing games""" if not shared.schema.get_boolean("remove-missing"): return @@ -249,7 +252,7 @@ class Importer(ErrorProducer): game.update() self.removed_game_ids.add(game.game_id) - def import_callback(self): + def import_callback(self) -> None: """Callback called when importing has finished""" logging.info("Import done") self.remove_games() @@ -261,11 +264,11 @@ class Importer(ErrorProducer): self.create_error_dialog() shared.win.get_application().lookup_action("import").set_enabled(True) - def create_error_dialog(self): + def create_error_dialog(self) -> None: """Dialog containing all errors raised by importers""" # Collect all errors that happened in the importer and the managers - errors: list[Exception] = [] + errors = [] errors.extend(self.collect_errors()) for manager in shared.store.managers.values(): errors.extend(manager.collect_errors()) @@ -313,7 +316,7 @@ class Importer(ErrorProducer): dialog.present() - def undo_import(self, *_args): + def undo_import(self, *_args: Any) -> None: for game_id in self.imported_game_ids: shared.store[game_id].removed = True shared.store[game_id].update() @@ -330,7 +333,7 @@ class Importer(ErrorProducer): logging.info("Import undone") - def create_summary_toast(self): + def create_summary_toast(self) -> Adw.Toast: """N games imported, removed toast""" toast = Adw.Toast(timeout=0, priority=Adw.ToastPriority.HIGH) @@ -371,16 +374,20 @@ class Importer(ErrorProducer): shared.win.toast_overlay.add_toast(toast) return toast - def open_preferences(self, page=None, expander_row=None): + def open_preferences( + self, + page_name: Optional[str] = None, + expander_row: Optional[Adw.ExpanderRow] = None, + ) -> Adw.PreferencesWindow: return shared.win.get_application().on_preferences_action( - page_name=page, expander_row=expander_row + page_name=page_name, expander_row=expander_row ) - def timeout_toast(self, *_args): + def timeout_toast(self, *_args: Any) -> None: """Manually timeout the toast after the user has dismissed all warnings""" GLib.timeout_add_seconds(5, self.summary_toast.dismiss) - def dialog_response_callback(self, _widget, response, *args): + def dialog_response_callback(self, _widget: Any, response: str, *args: Any) -> None: """Handle after-import dialogs callback""" logging.debug("After-import dialog response: %s (%s)", response, str(args)) if response == "open_preferences": diff --git a/src/importer/sources/location.py b/src/importer/sources/location.py index dde8e3f..36f592b 100644 --- a/src/importer/sources/location.py +++ b/src/importer/sources/location.py @@ -1,7 +1,7 @@ import logging from os import PathLike from pathlib import Path -from typing import Iterable, Mapping, NamedTuple +from typing import Iterable, Mapping, NamedTuple, Optional from src import shared @@ -41,7 +41,7 @@ class Location: paths: Mapping[str, LocationSubPath] invalid_subtitle: str - root: Path = None + root: Optional[Path] = None def __init__( self, @@ -94,7 +94,9 @@ class Location: shared.schema.set_string(self.schema_key, value) logging.debug("Resolved value for schema key %s: %s", self.schema_key, value) - def __getitem__(self, key: str): + def __getitem__(self, key: str) -> Optional[Path]: """Get the computed path from its key for the location""" self.resolve() - return self.root / self.paths[key].segment + if self.root: + return self.root / self.paths[key].segment + return None diff --git a/src/importer/sources/source.py b/src/importer/sources/source.py index 499881a..74a13eb 100644 --- a/src/importer/sources/source.py +++ b/src/importer/sources/source.py @@ -20,19 +20,19 @@ import sys from abc import abstractmethod from collections.abc import Iterable -from typing import Any, Collection, Generator +from typing import Any, Collection, Generator, Optional from src.game import Game from src.importer.sources.location import Location # Type of the data returned by iterating on a Source -SourceIterationResult = None | Game | tuple[Game, tuple[Any]] +SourceIterationResult = Optional[Game | tuple[Game, tuple[Any]]] class SourceIterable(Iterable): """Data producer for a source of games""" - source: "Source" = None + source: "Source" def __init__(self, source: "Source") -> None: self.source = source @@ -53,7 +53,7 @@ class Source(Iterable): source_id: str name: str - variant: str = None + variant: Optional[str] = None available_on: set[str] = set() iterable_class: type[SourceIterable] @@ -65,7 +65,7 @@ class Source(Iterable): def full_name(self) -> str: """The source's full name""" full_name_ = self.name - if self.variant is not None: + if self.variant: full_name_ += f" ({self.variant})" return full_name_ @@ -75,7 +75,7 @@ class Source(Iterable): return self.source_id + "_{game_id}" @property - def is_available(self): + def is_available(self) -> bool: return sys.platform in self.available_on @property diff --git a/src/logging/color_log_formatter.py b/src/logging/color_log_formatter.py index 53fe261..6f5545e 100644 --- a/src/logging/color_log_formatter.py +++ b/src/logging/color_log_formatter.py @@ -29,7 +29,7 @@ class ColorLogFormatter(Formatter): RED = "\033[31m" YELLOW = "\033[33m" - def format(self, record: LogRecord): + def format(self, record: LogRecord) -> str: super_format = super().format(record) match record.levelname: case "CRITICAL": diff --git a/src/logging/session_file_handler.py b/src/logging/session_file_handler.py index bc6e06e..a6f46b3 100644 --- a/src/logging/session_file_handler.py +++ b/src/logging/session_file_handler.py @@ -18,11 +18,12 @@ # SPDX-License-Identifier: GPL-3.0-or-later import lzma -from io import StringIO +from io import TextIOWrapper from logging import StreamHandler from lzma import FORMAT_XZ, PRESET_DEFAULT from os import PathLike from pathlib import Path +from typing import Optional from src import shared @@ -37,7 +38,7 @@ class SessionFileHandler(StreamHandler): backup_count: int filename: Path - log_file: StringIO = None + log_file: Optional[TextIOWrapper] = None def create_dir(self) -> None: """Create the log dir if needed""" @@ -83,7 +84,7 @@ class SessionFileHandler(StreamHandler): logfiles.sort(key=self.file_sort_key, reverse=True) return logfiles - def rotate_file(self, path: Path): + def rotate_file(self, path: Path) -> None: """Rotate a file's number suffix and remove it if it's too old""" # If uncompressed, compress @@ -128,5 +129,6 @@ class SessionFileHandler(StreamHandler): super().__init__(self.log_file) def close(self) -> None: - self.log_file.close() + if self.log_file: + self.log_file.close() super().close() diff --git a/src/logging/setup.py b/src/logging/setup.py index e9737cd..f3498cc 100644 --- a/src/logging/setup.py +++ b/src/logging/setup.py @@ -27,7 +27,7 @@ import sys from src import shared -def setup_logging(): +def setup_logging() -> None: """Intitate the app's logging""" is_dev = shared.PROFILE == "development" @@ -89,7 +89,7 @@ def setup_logging(): logging_dot_config.dictConfig(config) -def log_system_info(): +def log_system_info() -> None: """Log system debug information""" logging.debug("Starting %s v%s (%s)", shared.APP_ID, shared.VERSION, shared.PROFILE) diff --git a/src/main.py b/src/main.py index a1e3fdf..6c59d22 100644 --- a/src/main.py +++ b/src/main.py @@ -21,7 +21,7 @@ import json import lzma import os import sys -from typing import Any +from typing import Any, Optional import gi @@ -199,8 +199,8 @@ class CartridgesApplication(Adw.Application): self, _action: Any = None, _parameter: Any = None, - page_name: (str | None) = None, - expander_row: (str | None) = None, + page_name: Optional[str] = None, + expander_row: Optional[str] = None, ) -> CartridgesWindow: win = PreferencesWindow() if page_name: diff --git a/src/preferences.py b/src/preferences.py index c2f2596..9adf84b 100644 --- a/src/preferences.py +++ b/src/preferences.py @@ -21,7 +21,7 @@ import logging import re from pathlib import Path from shutil import rmtree -from typing import Any, Callable +from typing import Any, Callable, Optional from gi.repository import Adw, Gio, GLib, Gtk @@ -215,7 +215,7 @@ class PreferencesWindow(Adw.PreferencesWindow): ) def choose_folder( - self, _widget: Any, callback: Callable, callback_data: str | None = None + self, _widget: Any, callback: Callable, callback_data: Optional[str] = None ) -> None: self.file_chooser.select_folder(self.win, None, callback, callback_data) diff --git a/src/store/managers/manager.py b/src/store/managers/manager.py index 4ef50d0..a727a22 100644 --- a/src/store/managers/manager.py +++ b/src/store/managers/manager.py @@ -46,7 +46,7 @@ class Manager(ErrorProducer): max_tries: int = 3 @property - def name(self): + def name(self) -> str: return type(self).__name__ @abstractmethod @@ -59,13 +59,13 @@ class Manager(ErrorProducer): * May raise other exceptions that will be reported """ - def run(self, game: Game, additional_data: dict): + def run(self, game: Game, additional_data: dict) -> None: """Handle errors (retry, ignore or raise) that occur in the manager logic""" # Keep track of the number of tries tries = 1 - def handle_error(error: Exception): + def handle_error(error: Exception) -> None: nonlocal tries # If FriendlyError, handle its cause instead @@ -83,11 +83,11 @@ class Manager(ErrorProducer): retrying_format = "Retrying %s in %s for %s" unretryable_format = "Unretryable %s in %s for %s" - if error in self.continue_on: + if type(error) in self.continue_on: # Handle skippable errors (skip silently) return - if error in self.retryable_on: + if type(error) in self.retryable_on: if tries > self.max_tries: # Handle being out of retries logging.error(out_of_retries_format, *log_args) @@ -104,7 +104,7 @@ class Manager(ErrorProducer): logging.error(unretryable_format, *log_args, exc_info=error) self.report_error(base_error) - def try_manager_logic(): + def try_manager_logic() -> None: try: self.main(game, additional_data) except Exception as error: # pylint: disable=broad-exception-caught diff --git a/src/store/pipeline.py b/src/store/pipeline.py index 559ee7f..f552f04 100644 --- a/src/store/pipeline.py +++ b/src/store/pipeline.py @@ -83,7 +83,7 @@ class Pipeline(GObject.Object): progress = 1 return progress - def advance(self): + def advance(self) -> None: """Spawn tasks for managers that are able to run for a game""" # Separate blocking / async managers diff --git a/src/store/store.py b/src/store/store.py index a11bf5a..159da90 100644 --- a/src/store/store.py +++ b/src/store/store.py @@ -18,7 +18,7 @@ # SPDX-License-Identifier: GPL-3.0-or-later import logging -from typing import Any, Generator, MutableMapping +from typing import Any, Generator, MutableMapping, Optional from src import shared from src.game import Game @@ -77,13 +77,15 @@ class Store: except KeyError: return default - def add_manager(self, manager: Manager, in_pipeline=True): + def add_manager(self, manager: Manager, in_pipeline: bool = True) -> None: """Add a manager to the store""" manager_type = type(manager) self.managers[manager_type] = manager self.toggle_manager_in_pipelines(manager_type, in_pipeline) - def toggle_manager_in_pipelines(self, manager_type: type[Manager], enable: bool): + def toggle_manager_in_pipelines( + self, manager_type: type[Manager], enable: bool + ) -> None: """Change if a manager should run in new pipelines""" if enable: self.pipeline_managers.add(self.managers[manager_type]) @@ -108,8 +110,8 @@ class Store: pass def add_game( - self, game: Game, additional_data: dict, run_pipeline=True - ) -> Pipeline | None: + self, game: Game, additional_data: dict, run_pipeline: bool = True + ) -> Optional[Pipeline]: """Add a game to the app""" # Ignore games from a newer spec version diff --git a/src/window.py b/src/window.py index 6b485d8..b3af2f4 100644 --- a/src/window.py +++ b/src/window.py @@ -17,7 +17,7 @@ # # SPDX-License-Identifier: GPL-3.0-or-later -from typing import Any +from typing import Any, Optional from gi.repository import Adw, Gio, GLib, Gtk @@ -71,7 +71,7 @@ class CartridgesWindow(Adw.ApplicationWindow): game_covers: dict = {} toasts: dict = {} active_game: Game - details_view_game_cover: GameCover | None = None + details_view_game_cover: Optional[GameCover] = None sort_state: str = "a-z" def __init__(self, **kwargs: Any) -> None: @@ -334,7 +334,7 @@ class CartridgesWindow(Adw.ApplicationWindow): index += 1 def on_undo_action( - self, _widget: Any, game: Game | None = None, undo: str | None = None + self, _widget: Any, game: Optional[Game] = None, undo: Optional[str] = None ) -> None: if not game: # If the action was activated via Ctrl + Z if shared.importer and (