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

🎨 Better error handling in managers

Browse Source
This commit is contained in:
GeoffreyCoulaud
2023-05-31 17:21:01 +02:00
parent d204737339
commit ef63210a8f
4 changed files with 12 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/main.py
View File

@@ -247,7 +247,7 @@ def main(version): # pylint: disable=unused-argument
base_log_level = os.environ.get("LOGLEVEL", profile_base_log_level).upper() base_log_level = os.environ.get("LOGLEVEL", profile_base_log_level).upper()
lib_log_level = os.environ.get("LIBLOGLEVEL", profile_lib_log_level).upper() lib_log_level = os.environ.get("LIBLOGLEVEL", profile_lib_log_level).upper()
log_levels = { log_levels = {
__name__: base_log_level, None: base_log_level,
"PIL": lib_log_level, "PIL": lib_log_level,
"urllib3": lib_log_level, "urllib3": lib_log_level,
} }

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

@@ -56,22 +56,24 @@ class Manager:
self.manager_logic(game) self.manager_logic(game)
except Exception as error: except Exception as error:
# Handle unretryable errors # Handle unretryable errors
log_args = (type(error).__name__, self.name, game.game_id) log_args = (type(error).__name__, self.name, game.name, game.game_id)
if type(error) in self.retryable_on: if type(error) not in self.retryable_on:
logging.error("Unretryable %s in %s for %s", *log_args) logging.error("Unretryable %s in %s for %s (%s)", *log_args)
self.report_error(error) self.report_error(error)
break break
# Handle being out of retries # Handle being out of retries
elif remaining_tries == 0: elif remaining_tries == 0:
logging.error("Too many retries due to %s in %s for %s", *log_args) logging.error(
"Too many retries due to %s in %s for %s (%s)", *log_args
)
self.report_error(error) self.report_error(error)
break break
# Retry # Retry
else: else:
logging.debug("Retry caused by %s in %s for %s", *log_args) logging.debug("Retry caused by %s in %s for %s (%s)", *log_args)
continue continue
def process_game(self, game: Game, callback: Callable[["Manager"], Any]) -> None: def process_game(self, game: Game, callback: Callable[["Manager"], Any]) -> None:
"""Pass the game through the manager""" """Pass the game through the manager"""
self.execute_resilient_manager_logic(game, tries=0) self.execute_resilient_manager_logic(game)
callback(self) callback(self)

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

@@ -1,9 +1,7 @@
from requests import HTTPError
from src.game import Game from src.game import Game
from src.store.managers.async_manager import AsyncManager from src.store.managers.async_manager import AsyncManager
from src.store.managers.steam_api_manager import SteamAPIManager from src.store.managers.steam_api_manager import SteamAPIManager
from src.utils.steamgriddb import HTTPError, SGDBAuthError, SGDBError, SGDBHelper from src.utils.steamgriddb import HTTPError, SGDBAuthError, SGDBHelper
class SGDBManager(AsyncManager): class SGDBManager(AsyncManager):

6
src/utils/steam.py
View File

@@ -66,11 +66,9 @@ class SteamHelper:
) as response: ) as response:
response.raise_for_status() response.raise_for_status()
data = response.json()[appid] data = response.json()[appid]
except HTTPError as error:
except (HTTPError, JSONDecodeError) as error: logging.warning("Steam API HTTP error for %s", appid, exc_info=error)
logging.warning("Error while querying Steam API for %s", appid)
raise error raise error
if not data["success"]: if not data["success"]:
logging.debug("Appid %s not found", appid) logging.debug("Appid %s not found", appid)
raise SteamGameNotFoundError() raise SteamGameNotFoundError()