skullbite/cartridges
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

🎨 Work on import error handling

Browse Source 
- Generic ErrorProducer class
- Importer and managers are error producers
- SGDB Auth friendly error
- Bad source location friendly errors (data, config, cache)
- Removed unused decorators
This commit is contained in:
GeoffreyCoulaud
2023-06-24 15:13:35 +02:00
parent 41c2a1023a
commit 3fa80a53c6
10 changed files with 136 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
src/errors/error_producer.py Normal file
View File

@@ -0,0 +1,28 @@
from threading import Lock
class ErrorProducer:
"""
A mixin for objects that produce errors.
Specifies the report_error and collect_errors methods in a thread-safe manner.
"""
errors: list[Exception] = None
errors_lock: Lock = None
def __init__(self) -> None:
self.errors = []
self.errors_lock = Lock()
def report_error(self, error: Exception):
"""Report an error"""
with self.errors_lock:
self.errors.append(error)
def collect_errors(self) -> list[Exception]:
"""Collect and remove the errors produced by the object"""
with self.errors_lock:
errors = self.errors.copy()
self.errors.clear()
return errors

47
src/errors/friendly_error.py Normal file
View File

@@ -0,0 +1,47 @@
from typing import Iterable
class FriendlyError(Exception):
"""
An error that is supposed to be shown to the user in a nice format
Use `raise ... from ...` to preserve context.
"""
title_format: str
title_args: Iterable[str]
subtitle_format: str
subtitle_args: Iterable[str]
@property
def title(self) -> str:
"""Get the gettext translated error title"""
return _(self.title_format).format(self.title_args)
@property
def subtitle(self) -> str:
"""Get the gettext translated error subtitle"""
return _(self.subtitle_format).format(self.subtitle_args)
def __init__(
self,
title: str,
subtitle: str,
title_args: Iterable[str] = None,
subtitle_args: Iterable[str] = None,
) -> None:
"""Create a friendly error
:param str title: The error's title, translatable with gettext
:param str subtitle: The error's subtitle, translatable with gettext
"""
super().__init__()
if title is not None:
self.title_format = title
if subtitle is not None:
self.subtitle_format = subtitle
self.title_args = title_args if title_args else ()
self.subtitle_args = subtitle_args if subtitle_args else ()
def __str__(self) -> str:
return f"{self.title} - {self.subtitle}"

35
src/importer/importer.py
View File

@@ -20,18 +20,21 @@
import logging
from gi.repository import Adw, Gtk, GLib
from gi.repository import Adw, GLib, Gtk
from src import shared
from src.errors.error_producer import ErrorProducer
from src.errors.friendly_error import FriendlyError
from src.game import Game
from src.importer.sources.source import Source
from src.store.managers.async_manager import AsyncManager
from src.store.pipeline import Pipeline
from src.utils.create_dialog import create_dialog
from src.utils.task import Task
# pylint: disable=too-many-instance-attributes
class Importer:
class Importer(ErrorProducer):
"""A class in charge of scanning sources for games"""
progressbar = None
@@ -47,6 +50,7 @@ class Importer:
game_pipelines: set[Pipeline] = None
def __init__(self):
super().__init__()
self.game_pipelines = set()
self.sources = set()
@@ -81,6 +85,15 @@ class Importer:
self.create_dialog()
# Collect all errors and reset the cancellables for the managers
# - Only one importer exists at any given time
# - Every import starts fresh
self.collect_errors()
for manager in shared.store.managers.values():
manager.collect_errors()
if isinstance(manager, AsyncManager):
manager.reset_cancellable()
for source in self.sources:
logging.debug("Importing games from source %s", source.id)
task = Task.new(None, None, self.source_callback, (source,))
@@ -129,10 +142,9 @@ class Importer:
iteration_result = next(iterator)
except StopIteration:
break
except Exception as exception: # pylint: disable=broad-exception-caught
logging.exception(
"Exception in source %s", source.id, exc_info=exception
)
except Exception as error: # pylint: disable=broad-exception-caught
logging.exception("%s in %s", type(error).__name__, source.id)
self.report_error(error)
continue
# Handle the result depending on its type
@@ -202,9 +214,16 @@ class Importer:
string = _("The following errors occured during import:")
errors = ""
# Collect all errors that happened in the importer and the managers
collected_errors: list[Exception] = []
collected_errors.extend(self.collect_errors())
for manager in shared.store.managers.values():
for error in manager.collect_errors():
errors += "\n\n" + str(error)
collected_errors.extend(manager.collect_errors())
for error in collected_errors:
# Only display friendly errors
if not isinstance(error, FriendlyError):
continue
errors += "\n\n" + str(error)
if errors:
create_dialog(

1
src/importer/sources/legendary_source.py
View File

@@ -20,7 +20,6 @@
import json
import logging
from json import JSONDecodeError
from pathlib import Path
from time import time
from typing import Generator

20
src/importer/sources/source.py
View File

@@ -20,10 +20,11 @@
import sys
from abc import abstractmethod
from collections.abc import Iterable, Iterator
from typing import Generator, Any, Optional
from typing import Any, Generator, Optional
from src.importer.sources.location import Location
from src.errors.friendly_error import FriendlyError
from src.game import Game
from src.importer.sources.location import Location, UnresolvableLocationError
# Type of the data returned by iterating on a Source
SourceIterationResult = None | Game | tuple[Game, tuple[Any]]
@@ -100,9 +101,20 @@ class Source(Iterable):
def __iter__(self) -> SourceIterator:
"""Get an iterator for the source"""
for location in (self.data_location, self.cache_location, self.config_location):
if location is not None:
for location_name in ("data", "cache", "config"):
location = getattr(self, f"{location_name}_location", None)
if location is None:
continue
try:
location.resolve()
except UnresolvableLocationError as error:
raise FriendlyError(
# The variable is the source's name
f"Invalid {location_name} location for {{}}",
"Change it or disable the source in the preferences",
(self.name,),
(self.name,),
) from error
return self.iterator_class(self)

4
src/importer/sources/steam_source.py
View File

@@ -30,10 +30,6 @@ from src.importer.sources.source import (
SourceIterator,
URLExecutableSource,
)
from src.utils.decorators import (
replaced_by_path,
replaced_by_schema_key,
)
from src.utils.steam import SteamFileHelper, SteamInvalidManifestError
from src.importer.sources.location import Location

1
src/meson.build
View File

@@ -12,6 +12,7 @@ install_subdir('importer', install_dir: moduledir)
install_subdir('utils', install_dir: moduledir)
install_subdir('store', install_dir: moduledir)
install_subdir('logging', install_dir: moduledir)
install_subdir('errors', install_dir: moduledir)
install_data(
[
'main.py',

34
src/store/managers/manager.py
View File

@@ -19,14 +19,15 @@
import logging
from abc import abstractmethod
from threading import Lock
from time import sleep
from typing import Any, Callable, Container
from src.errors.error_producer import ErrorProducer
from src.errors.friendly_error import FriendlyError
from src.game import Game
class Manager:
class Manager(ErrorProducer):
"""Class in charge of handling a post creation action for games.
* May connect to signals on the game to handle them.
@@ -44,30 +45,10 @@ class Manager:
retry_delay: int = 3
max_tries: int = 3
errors: list[Exception]
errors_lock: Lock = None
@property
def name(self):
return type(self).__name__
def __init__(self) -> None:
super().__init__()
self.errors = []
self.errors_lock = Lock()
def report_error(self, error: Exception):
"""Report an error that happened in Manager.process_game"""
with self.errors_lock:
self.errors.append(error)
def collect_errors(self) -> list[Exception]:
"""Get the errors produced by the manager and remove them from self.errors"""
with self.errors_lock:
errors = self.errors.copy()
self.errors.clear()
return errors
@abstractmethod
def manager_logic(self, game: Game, additional_data: dict) -> None:
"""
@@ -87,6 +68,11 @@ class Manager:
def handle_error(error: Exception):
nonlocal tries
# If FriendlyError, handle its cause instead
base_error = error
if isinstance(error, FriendlyError):
error = error.__cause__
log_args = (
type(error).__name__,
self.name,
@@ -105,7 +91,7 @@ class Manager:
if tries > self.max_tries:
# Handle being out of retries
logging.error(out_of_retries_format, *log_args)
self.report_error(error)
self.report_error(base_error)
else:
# Handle retryable errors
logging.error(retrying_format, *log_args)
@@ -116,7 +102,7 @@ class Manager:
else:
# Handle unretryable errors
logging.error(unretryable_format, *log_args, exc_info=error)
self.report_error(error)
self.report_error(base_error)
def try_manager_logic():
try:

10
src/store/managers/sgdb_manager.py
View File

@@ -21,10 +21,11 @@ from json import JSONDecodeError
from requests.exceptions import HTTPError, SSLError
from src.errors.friendly_error import FriendlyError
from src.game import Game
from src.store.managers.async_manager import AsyncManager
from src.store.managers.online_cover_manager import OnlineCoverManager
from src.store.managers.local_cover_manager import LocalCoverManager
from src.store.managers.online_cover_manager import OnlineCoverManager
from src.store.managers.steam_api_manager import SteamAPIManager
from src.utils.steamgriddb import SGDBAuthError, SGDBHelper
@@ -39,7 +40,10 @@ class SGDBManager(AsyncManager):
try:
sgdb = SGDBHelper()
sgdb.conditionaly_update_cover(game)
except SGDBAuthError:
except SGDBAuthError as error:
# If invalid auth, cancel all SGDBManager tasks
self.cancellable.cancel()
raise
raise FriendlyError(
"Couldn't authenticate to SGDB",
"Verify your API key in the preferences",
) from error

56
src/utils/decorators.py
View File

@@ -1,56 +0,0 @@
# decorators.py
#
# Copyright 2023 Geoffrey Coulaud
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# SPDX-License-Identifier: GPL-3.0-or-later
from pathlib import Path
from os import PathLike
from functools import wraps
from src import shared
def replaced_by_path(override: PathLike): # Decorator builder
"""Replace the method's returned path with the override
if the override exists on disk"""
def decorator(original_function): # Built decorator (closure)
@wraps(original_function)
def wrapper(*args, **kwargs): # func's override
path = Path(override).expanduser()
if path.exists():
return path
return original_function(*args, **kwargs)
return wrapper
return decorator
def replaced_by_schema_key(original_method): # Built decorator (closure)
"""
Replace the original method's value by the path pointed at in the schema
by the class' location key (if that override exists)
"""
@wraps(original_method)
def wrapper(*args, **kwargs): # func's override
source = args[0]
override = shared.schema.get_string(source.location_key)
return replaced_by_path(override)(original_method)(*args, **kwargs)
return wrapper