rommapp/romm
1
0
Fork 0
mirror of https://github.com/rommapp/romm.git synced 2026-02-18 00:27:41 +01:00
Code Issues Packages Projects Releases Wiki Activity

[ROMM-2657] Safe access env vars with defaults

Browse Source
This commit is contained in:
Georges-Antoine Assi
2025-11-15 14:54:21 -05:00
parent a5f414176a
commit 5c0bd9c418
8 changed files with 139 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files

239
backend/config/__init__.py
View File

@@ -1,219 +1,220 @@
import os
from typing import Final
from typing import Final, overload
import yarl
from dotenv import load_dotenv
from utils.database import safe_int, safe_str_to_bool
load_dotenv()
def str_to_bool(value: str) -> bool:
return value.strip().lower() in ("1", "true", "yes", "on")
# Supplying a string literal for fallback guarantees a `str` result
@overload
def _get_env(var: str, fallback: str) -> str: ...
@overload
def _get_env(var: str, fallback: None = None) -> str | None: ...
ROMM_BASE_URL = os.environ.get("ROMM_BASE_URL", "http://0.0.0.0")
ROMM_PORT = int(os.environ.get("ROMM_PORT", 8080))
def _get_env(var: str, fallback: str | None = None) -> str | None:
return os.environ.get(var) or fallback
ROMM_BASE_URL: Final[str] = _get_env("ROMM_BASE_URL", "http://0.0.0.0")
ROMM_PORT: Final[int] = safe_int(_get_env("ROMM_PORT"), 8080)
# GUNICORN
DEV_MODE: Final = str_to_bool(os.environ.get("DEV_MODE", "false"))
DEV_HOST: Final = os.environ.get("DEV_HOST", "127.0.0.1")
DEV_PORT: Final = int(os.environ.get("DEV_PORT", "5000"))
DEV_SQL_ECHO: Final = str_to_bool(os.environ.get("DEV_SQL_ECHO", "false"))
DEV_MODE: Final[bool] = safe_str_to_bool(_get_env("DEV_MODE"))
DEV_HOST: Final[str] = _get_env("DEV_HOST", "127.0.0.1")
DEV_PORT: Final[int] = safe_int(_get_env("DEV_PORT"), 5000)
DEV_SQL_ECHO: Final[bool] = safe_str_to_bool(_get_env("DEV_SQL_ECHO"))
# PATHS
ROMM_BASE_PATH: Final = os.environ.get("ROMM_BASE_PATH", "/romm")
ROMM_TMP_PATH: Final = os.environ.get("ROMM_TMP_PATH", None)
LIBRARY_BASE_PATH: Final = f"{ROMM_BASE_PATH}/library"
RESOURCES_BASE_PATH: Final = f"{ROMM_BASE_PATH}/resources"
ASSETS_BASE_PATH: Final = f"{ROMM_BASE_PATH}/assets"
FRONTEND_RESOURCES_PATH: Final = "/assets/romm/resources"
ROMM_BASE_PATH: Final[str] = _get_env("ROMM_BASE_PATH", "/romm")
ROMM_TMP_PATH: Final[str | None] = _get_env("ROMM_TMP_PATH")
LIBRARY_BASE_PATH: Final[str] = f"{ROMM_BASE_PATH}/library"
RESOURCES_BASE_PATH: Final[str] = f"{ROMM_BASE_PATH}/resources"
ASSETS_BASE_PATH: Final[str] = f"{ROMM_BASE_PATH}/assets"
FRONTEND_RESOURCES_PATH: Final[str] = "/assets/romm/resources"
# SEVEN ZIP
SEVEN_ZIP_TIMEOUT: Final = int(os.environ.get("SEVEN_ZIP_TIMEOUT", 60))
SEVEN_ZIP_TIMEOUT: Final[int] = safe_int(_get_env("SEVEN_ZIP_TIMEOUT"), 60)
# DATABASE
DB_HOST: Final[str | None] = os.environ.get("DB_HOST", "127.0.0.1") or None
DB_PORT: Final[int | None] = (
int(os.environ.get("DB_PORT", 3306)) if os.environ.get("DB_PORT") != "" else None
)
DB_USER: Final[str | None] = os.environ.get("DB_USER")
DB_PASSWD: Final[str | None] = os.environ.get("DB_PASSWD")
DB_NAME: Final[str] = os.environ.get("DB_NAME", "romm")
DB_QUERY_JSON: Final[str | None] = os.environ.get("DB_QUERY_JSON")
ROMM_DB_DRIVER: Final[str] = os.environ.get("ROMM_DB_DRIVER", "mariadb")
DB_HOST: Final[str] = _get_env("DB_HOST", "127.0.0.1")
DB_PORT: Final[int] = safe_int(_get_env("DB_PORT", "3306"))
DB_USER: Final[str | None] = _get_env("DB_USER")
DB_PASSWD: Final[str | None] = _get_env("DB_PASSWD")
DB_NAME: Final[str] = _get_env("DB_NAME", "romm")
DB_QUERY_JSON: Final[str | None] = _get_env("DB_QUERY_JSON")
ROMM_DB_DRIVER: Final[str] = _get_env("ROMM_DB_DRIVER", "mariadb")
# REDIS
REDIS_HOST: Final = os.environ.get("REDIS_HOST", "127.0.0.1")
REDIS_PORT: Final = int(os.environ.get("REDIS_PORT", 6379))
REDIS_PASSWORD: Final = os.environ.get("REDIS_PASSWORD")
REDIS_USERNAME: Final = os.environ.get("REDIS_USERNAME", "")
REDIS_DB: Final = int(os.environ.get("REDIS_DB", 0))
REDIS_SSL: Final = str_to_bool(os.environ.get("REDIS_SSL", "false"))
REDIS_URL: Final = yarl.URL.build(
scheme="rediss" if REDIS_SSL else "redis",
user=REDIS_USERNAME or None,
password=REDIS_PASSWORD or None,
host=REDIS_HOST,
port=REDIS_PORT,
path=f"/{REDIS_DB}",
REDIS_HOST: Final[str] = _get_env("REDIS_HOST", "127.0.0.1")
REDIS_PORT: Final[int] = safe_int(_get_env("REDIS_PORT"), 6379)
REDIS_PASSWORD: Final[str | None] = _get_env("REDIS_PASSWORD")
REDIS_USERNAME: Final[str | None] = _get_env("REDIS_USERNAME", "")
REDIS_DB: Final[int] = safe_int(_get_env("REDIS_DB"), 0)
REDIS_SSL: Final[bool] = safe_str_to_bool(_get_env("REDIS_SSL"))
REDIS_URL: Final[str] = str(
yarl.URL.build(
scheme="rediss" if REDIS_SSL else "redis",
user=REDIS_USERNAME or None,
password=REDIS_PASSWORD or None,
host=REDIS_HOST,
port=REDIS_PORT,
path=f"/{REDIS_DB}",
)
)
# IGDB
IGDB_CLIENT_ID: Final[str] = os.environ.get(
"IGDB_CLIENT_ID", os.environ.get("CLIENT_ID", "")
).strip()
IGDB_CLIENT_SECRET: Final[str] = os.environ.get(
"IGDB_CLIENT_SECRET", os.environ.get("CLIENT_SECRET", "")
).strip()
IGDB_CLIENT_ID: Final[str | None] = _get_env("IGDB_CLIENT_ID")
IGDB_CLIENT_SECRET: Final[str | None] = _get_env("IGDB_CLIENT_SECRET")
# MOBYGAMES
MOBYGAMES_API_KEY: Final[str] = os.environ.get("MOBYGAMES_API_KEY", "").strip()
MOBYGAMES_API_KEY: Final[str | None] = _get_env("MOBYGAMES_API_KEY")
# SCREENSCRAPER
SCREENSCRAPER_USER: Final[str] = os.environ.get("SCREENSCRAPER_USER", "")
SCREENSCRAPER_PASSWORD: Final[str] = os.environ.get("SCREENSCRAPER_PASSWORD", "")
SCREENSCRAPER_USER: Final[str | None] = _get_env("SCREENSCRAPER_USER")
SCREENSCRAPER_PASSWORD: Final[str | None] = _get_env("SCREENSCRAPER_PASSWORD")
# STEAMGRIDDB
STEAMGRIDDB_API_KEY: Final[str] = os.environ.get("STEAMGRIDDB_API_KEY", "").strip()
STEAMGRIDDB_API_KEY: Final[str | None] = _get_env("STEAMGRIDDB_API_KEY")
# RETROACHIEVEMENTS
RETROACHIEVEMENTS_API_KEY: Final[str] = os.environ.get("RETROACHIEVEMENTS_API_KEY", "")
REFRESH_RETROACHIEVEMENTS_CACHE_DAYS: Final[int] = int(
os.environ.get("REFRESH_RETROACHIEVEMENTS_CACHE_DAYS", 30)
RETROACHIEVEMENTS_API_KEY: Final[str | None] = _get_env("RETROACHIEVEMENTS_API_KEY")
REFRESH_RETROACHIEVEMENTS_CACHE_DAYS: Final[int] = safe_int(
_get_env("REFRESH_RETROACHIEVEMENTS_CACHE_DAYS"), 30
)
# LAUNCHBOX
LAUNCHBOX_API_ENABLED: Final[bool] = str_to_bool(
os.environ.get("LAUNCHBOX_API_ENABLED", "false")
)
LAUNCHBOX_API_ENABLED: Final[bool] = safe_str_to_bool(_get_env("LAUNCHBOX_API_ENABLED"))
# PLAYMATCH
PLAYMATCH_API_ENABLED: Final[bool] = str_to_bool(
os.environ.get("PLAYMATCH_API_ENABLED", "false")
)
PLAYMATCH_API_ENABLED: Final[bool] = safe_str_to_bool(_get_env("PLAYMATCH_API_ENABLED"))
# HASHEOUS
HASHEOUS_API_ENABLED: Final[bool] = str_to_bool(
os.environ.get("HASHEOUS_API_ENABLED", "false")
)
HASHEOUS_API_ENABLED: Final[bool] = safe_str_to_bool(_get_env("HASHEOUS_API_ENABLED"))
# THEGAMESDB
TGDB_API_ENABLED: Final[bool] = str_to_bool(os.environ.get("TGDB_API_ENABLED", "false"))
TGDB_API_ENABLED: Final[bool] = safe_str_to_bool(_get_env("TGDB_API_ENABLED"))
# FLASHPOINT
FLASHPOINT_API_ENABLED: Final = str_to_bool(
os.environ.get("FLASHPOINT_API_ENABLED", "false")
FLASHPOINT_API_ENABLED: Final[bool] = safe_str_to_bool(
_get_env("FLASHPOINT_API_ENABLED")
)
# HOWLONGTOBEAT
HLTB_API_ENABLED: Final = str_to_bool(os.environ.get("HLTB_API_ENABLED", "false"))
HLTB_API_ENABLED: Final[bool] = safe_str_to_bool(_get_env("HLTB_API_ENABLED"))
# AUTH
ROMM_AUTH_SECRET_KEY: Final[str] = os.environ.get("ROMM_AUTH_SECRET_KEY", "")
ROMM_AUTH_SECRET_KEY: Final[str | None] = _get_env("ROMM_AUTH_SECRET_KEY")
if not ROMM_AUTH_SECRET_KEY:
raise ValueError("ROMM_AUTH_SECRET_KEY environment variable is not set!")
SESSION_MAX_AGE_SECONDS: Final = int(
os.environ.get("SESSION_MAX_AGE_SECONDS", 14 * 24 * 60 * 60)
SESSION_MAX_AGE_SECONDS: Final[int] = safe_int(
_get_env("SESSION_MAX_AGE_SECONDS"), 14 * 24 * 60 * 60
) # 14 days, in seconds
DISABLE_CSRF_PROTECTION = str_to_bool(
os.environ.get("DISABLE_CSRF_PROTECTION", "false")
DISABLE_CSRF_PROTECTION: Final[bool] = safe_str_to_bool(
_get_env("DISABLE_CSRF_PROTECTION")
)
DISABLE_DOWNLOAD_ENDPOINT_AUTH = str_to_bool(
os.environ.get("DISABLE_DOWNLOAD_ENDPOINT_AUTH", "false")
DISABLE_DOWNLOAD_ENDPOINT_AUTH: Final[bool] = safe_str_to_bool(
_get_env("DISABLE_DOWNLOAD_ENDPOINT_AUTH")
)
DISABLE_USERPASS_LOGIN = str_to_bool(os.environ.get("DISABLE_USERPASS_LOGIN", "false"))
DISABLE_SETUP_WIZARD = str_to_bool(os.environ.get("DISABLE_SETUP_WIZARD", "false"))
DISABLE_USERPASS_LOGIN: Final[bool] = safe_str_to_bool(
_get_env("DISABLE_USERPASS_LOGIN")
)
DISABLE_SETUP_WIZARD: Final[bool] = safe_str_to_bool(_get_env("DISABLE_SETUP_WIZARD"))
# OIDC
OIDC_ENABLED: Final = str_to_bool(os.environ.get("OIDC_ENABLED", "false"))
OIDC_PROVIDER: Final = os.environ.get("OIDC_PROVIDER", "")
OIDC_CLIENT_ID: Final = os.environ.get("OIDC_CLIENT_ID", "").strip()
OIDC_CLIENT_SECRET: Final = os.environ.get("OIDC_CLIENT_SECRET", "").strip()
OIDC_CLAIM_ROLES: Final = os.environ.get("OIDC_CLAIM_ROLES", "").strip()
OIDC_ROLE_VIEWER: Final = os.environ.get("OIDC_ROLE_VIEWER", "").strip()
OIDC_ROLE_EDITOR: Final = os.environ.get("OIDC_ROLE_EDITOR", "").strip()
OIDC_ROLE_ADMIN: Final = os.environ.get("OIDC_ROLE_ADMIN", "").strip()
OIDC_REDIRECT_URI: Final = os.environ.get("OIDC_REDIRECT_URI", "")
OIDC_SERVER_APPLICATION_URL: Final = os.environ.get("OIDC_SERVER_APPLICATION_URL", "")
OIDC_TLS_CACERTFILE: Final = os.environ.get("OIDC_TLS_CACERTFILE", None)
OIDC_ENABLED: Final[bool] = safe_str_to_bool(_get_env("OIDC_ENABLED"))
OIDC_PROVIDER: Final[str | None] = _get_env("OIDC_PROVIDER")
OIDC_CLIENT_ID: Final[str | None] = _get_env("OIDC_CLIENT_ID")
OIDC_CLIENT_SECRET: Final[str | None] = _get_env("OIDC_CLIENT_SECRET")
OIDC_CLAIM_ROLES: Final[str | None] = _get_env("OIDC_CLAIM_ROLES")
OIDC_ROLE_VIEWER: Final[str | None] = _get_env("OIDC_ROLE_VIEWER")
OIDC_ROLE_EDITOR: Final[str | None] = _get_env("OIDC_ROLE_EDITOR")
OIDC_ROLE_ADMIN: Final[str | None] = _get_env("OIDC_ROLE_ADMIN")
OIDC_REDIRECT_URI: Final[str | None] = _get_env("OIDC_REDIRECT_URI")
OIDC_SERVER_APPLICATION_URL: Final[str | None] = _get_env("OIDC_SERVER_APPLICATION_URL")
OIDC_TLS_CACERTFILE: Final[str | None] = _get_env("OIDC_TLS_CACERTFILE", None)
# SCANS
SCAN_TIMEOUT: Final = int(os.environ.get("SCAN_TIMEOUT", 60 * 60 * 4)) # 4 hours
SCAN_WORKERS: Final = max(1, int(os.environ.get("SCAN_WORKERS", "1")))
SCAN_TIMEOUT: Final[int] = safe_int(_get_env("SCAN_TIMEOUT"), 60 * 60 * 4) # 4 hours
SCAN_WORKERS: Final[int] = max(1, safe_int(_get_env("SCAN_WORKERS"), 1))
# TASKS
TASK_TIMEOUT: Final = int(os.environ.get("TASK_TIMEOUT", 60 * 5)) # 5 minutes
TASK_RESULT_TTL: Final = int(
os.environ.get("TASK_RESULT_TTL", 24 * 60 * 60)
TASK_TIMEOUT: Final[int] = safe_int(_get_env("TASK_TIMEOUT"), 60 * 5) # 5 minutes
TASK_RESULT_TTL: Final[int] = safe_int(
_get_env("TASK_RESULT_TTL"), 24 * 60 * 60
) # 24 hours
ENABLE_RESCAN_ON_FILESYSTEM_CHANGE: Final = str_to_bool(
os.environ.get("ENABLE_RESCAN_ON_FILESYSTEM_CHANGE", "false")
ENABLE_RESCAN_ON_FILESYSTEM_CHANGE: Final[bool] = safe_str_to_bool(
_get_env("ENABLE_RESCAN_ON_FILESYSTEM_CHANGE")
)
RESCAN_ON_FILESYSTEM_CHANGE_DELAY: Final = int(
os.environ.get("RESCAN_ON_FILESYSTEM_CHANGE_DELAY", 5) # 5 minutes
RESCAN_ON_FILESYSTEM_CHANGE_DELAY: Final[int] = safe_int(
_get_env("RESCAN_ON_FILESYSTEM_CHANGE_DELAY"), 5 # 5 minutes
)
ENABLE_SCHEDULED_RESCAN: Final = str_to_bool(
os.environ.get("ENABLE_SCHEDULED_RESCAN", "false")
ENABLE_SCHEDULED_RESCAN: Final[bool] = safe_str_to_bool(
_get_env("ENABLE_SCHEDULED_RESCAN")
)
SCHEDULED_RESCAN_CRON: Final = os.environ.get(
SCHEDULED_RESCAN_CRON: Final[str] = _get_env(
"SCHEDULED_RESCAN_CRON",
"0 3 * * *", # At 3:00 AM every day
)
ENABLE_SCHEDULED_UPDATE_SWITCH_TITLEDB: Final = str_to_bool(
os.environ.get("ENABLE_SCHEDULED_UPDATE_SWITCH_TITLEDB", "false")
ENABLE_SCHEDULED_UPDATE_SWITCH_TITLEDB: Final[bool] = safe_str_to_bool(
_get_env("ENABLE_SCHEDULED_UPDATE_SWITCH_TITLEDB")
)
SCHEDULED_UPDATE_SWITCH_TITLEDB_CRON: Final = os.environ.get(
SCHEDULED_UPDATE_SWITCH_TITLEDB_CRON: Final[str] = _get_env(
"SCHEDULED_UPDATE_SWITCH_TITLEDB_CRON",
"0 4 * * *", # At 4:00 AM every day
)
ENABLE_SCHEDULED_UPDATE_LAUNCHBOX_METADATA: Final = str_to_bool(
os.environ.get("ENABLE_SCHEDULED_UPDATE_LAUNCHBOX_METADATA", "false")
ENABLE_SCHEDULED_UPDATE_LAUNCHBOX_METADATA: Final[bool] = safe_str_to_bool(
_get_env("ENABLE_SCHEDULED_UPDATE_LAUNCHBOX_METADATA")
)
SCHEDULED_UPDATE_LAUNCHBOX_METADATA_CRON: Final = os.environ.get(
SCHEDULED_UPDATE_LAUNCHBOX_METADATA_CRON: Final[str] = _get_env(
"SCHEDULED_UPDATE_LAUNCHBOX_METADATA_CRON",
"0 4 * * *", # At 4:00 AM every day
)
ENABLE_SCHEDULED_CONVERT_IMAGES_TO_WEBP: Final = str_to_bool(
os.environ.get("ENABLE_SCHEDULED_CONVERT_IMAGES_TO_WEBP", "false")
ENABLE_SCHEDULED_CONVERT_IMAGES_TO_WEBP: Final[bool] = safe_str_to_bool(
_get_env("ENABLE_SCHEDULED_CONVERT_IMAGES_TO_WEBP")
)
SCHEDULED_CONVERT_IMAGES_TO_WEBP_CRON: Final = os.environ.get(
SCHEDULED_CONVERT_IMAGES_TO_WEBP_CRON: Final[str] = _get_env(
"SCHEDULED_CONVERT_IMAGES_TO_WEBP_CRON",
"0 4 * * *", # At 4:00 AM every day
)
ENABLE_SCHEDULED_RETROACHIEVEMENTS_PROGRESS_SYNC: Final[bool] = str_to_bool(
os.environ.get("ENABLE_SCHEDULED_RETROACHIEVEMENTS_PROGRESS_SYNC", "false")
ENABLE_SCHEDULED_RETROACHIEVEMENTS_PROGRESS_SYNC: Final[bool] = safe_str_to_bool(
_get_env("ENABLE_SCHEDULED_RETROACHIEVEMENTS_PROGRESS_SYNC")
)
SCHEDULED_RETROACHIEVEMENTS_PROGRESS_SYNC_CRON: Final[str] = os.environ.get(
SCHEDULED_RETROACHIEVEMENTS_PROGRESS_SYNC_CRON: Final[str] = _get_env(
"SCHEDULED_RETROACHIEVEMENTS_PROGRESS_SYNC_CRON",
"0 4 * * *", # At 4:00 AM every day
)
# EMULATION
DISABLE_EMULATOR_JS = str_to_bool(os.environ.get("DISABLE_EMULATOR_JS", "false"))
DISABLE_RUFFLE_RS = str_to_bool(os.environ.get("DISABLE_RUFFLE_RS", "false"))
DISABLE_EMULATOR_JS: Final[bool] = safe_str_to_bool(_get_env("DISABLE_EMULATOR_JS"))
DISABLE_RUFFLE_RS: Final[bool] = safe_str_to_bool(_get_env("DISABLE_RUFFLE_RS"))
# FRONTEND
UPLOAD_TIMEOUT = int(os.environ.get("UPLOAD_TIMEOUT", 600))
KIOSK_MODE = str_to_bool(os.environ.get("KIOSK_MODE", "false"))
UPLOAD_TIMEOUT: Final[int] = safe_int(_get_env("UPLOAD_TIMEOUT"), 600)
KIOSK_MODE: Final[bool] = safe_str_to_bool(_get_env("KIOSK_MODE"))
# LOGGING
LOGLEVEL: Final = os.environ.get("LOGLEVEL", "INFO").upper()
FORCE_COLOR: Final = str_to_bool(os.environ.get("FORCE_COLOR", "false"))
NO_COLOR: Final = str_to_bool(os.environ.get("NO_COLOR", "false"))
LOGLEVEL: Final[str] = _get_env("LOGLEVEL", "INFO").upper()
FORCE_COLOR: Final[bool] = safe_str_to_bool(_get_env("FORCE_COLOR"))
NO_COLOR: Final[bool] = safe_str_to_bool(_get_env("NO_COLOR"))
# YOUTUBE
YOUTUBE_BASE_URL: Final = os.environ.get(
YOUTUBE_BASE_URL: Final[str] = _get_env(
"YOUTUBE_BASE_URL", "https://www.youtube.com"
).rstrip("/")
# TINFOIL
TINFOIL_WELCOME_MESSAGE: Final = os.environ.get(
TINFOIL_WELCOME_MESSAGE: Final[str] = _get_env(
"TINFOIL_WELCOME_MESSAGE", "RomM Switch Library"
)
# SENTRY
SENTRY_DSN: Final = os.environ.get("SENTRY_DSN", None)
SENTRY_DSN: Final[str | None] = _get_env("SENTRY_DSN")
# TESTING
IS_PYTEST_RUN: Final = bool(os.environ.get("PYTEST_VERSION", False))
IS_PYTEST_RUN: Final = bool(_get_env("PYTEST_VERSION"))

2
backend/decorators/auth.py
View File

@@ -48,7 +48,7 @@ config = Config(
"OIDC_CLIENT_SECRET": OIDC_CLIENT_SECRET,
"OIDC_REDIRECT_URI": OIDC_REDIRECT_URI,
"OIDC_SERVER_APPLICATION_URL": OIDC_SERVER_APPLICATION_URL,
}
} # type: ignore
)
oauth = OAuth(config=config)
oauth.register(

5
backend/endpoints/rom.py
View File

@@ -36,7 +36,6 @@ from config import (
DEV_MODE,
DISABLE_DOWNLOAD_ENDPOINT_AUTH,
LIBRARY_BASE_PATH,
str_to_bool,
)
from decorators.auth import protected_route
from endpoints.responses import BulkOperationResponse
@@ -66,7 +65,7 @@ from logger.formatter import BLUE
from logger.formatter import highlight as hl
from logger.logger import log
from models.rom import Rom
from utils.database import safe_int
from utils.database import safe_int, safe_str_to_bool
from utils.filesystem import sanitize_filename
from utils.hashing import crc32_to_hex
from utils.nginx import FileRedirectResponse, ZipContentLine, ZipResponse
@@ -567,7 +566,7 @@ async def get_rom_content(
raise RomNotFoundInDatabaseException(id)
# https://muos.dev/help/addcontent#what-about-multi-disc-content
hidden_folder = str_to_bool(request.query_params.get("hidden_folder", ""))
hidden_folder = safe_str_to_bool(request.query_params.get("hidden_folder", ""))
files = list(db_rom_handler.get_rom_files(rom.id))
if file_ids:

2
backend/endpoints/sockets/scan.py
View File

@@ -102,7 +102,7 @@ class ScanStats:
def _get_socket_manager() -> socketio.AsyncRedisManager:
"""Connect to external socketio server"""
return socketio.AsyncRedisManager(str(REDIS_URL), write_only=True)
return socketio.AsyncRedisManager(REDIS_URL, write_only=True)
async def _identify_firmware(

5
backend/handler/metadata/launchbox_handler.py
View File

@@ -5,9 +5,10 @@ from typing import Final, NotRequired, TypedDict
import pydash
from config import LAUNCHBOX_API_ENABLED, str_to_bool
from config import LAUNCHBOX_API_ENABLED
from handler.redis_handler import async_cache
from logger.logger import log
from utils.database import safe_str_to_bool
from .base_handler import BaseRom, MetadataHandler
from .base_handler import UniversalPlatformSlug as UPS
@@ -91,7 +92,7 @@ def extract_metadata_from_launchbox_rom(
"first_release_date": first_release_date,
"max_players": int(index_entry.get("MaxPlayers") or 0),
"release_type": index_entry.get("ReleaseType", ""),
"cooperative": str_to_bool(index_entry.get("Cooperative") or "false"),
"cooperative": safe_str_to_bool(index_entry.get("Cooperative") or "false"),
"youtube_video_id": extract_video_id_from_youtube_url(
index_entry.get("VideoURL")
),

6
backend/handler/redis_handler.py
View File

@@ -18,7 +18,7 @@ class QueuePrio(Enum):
LOW = "low"
redis_client = Redis.from_url(str(REDIS_URL))
redis_client = Redis.from_url(REDIS_URL)
high_prio_queue = Queue(name=QueuePrio.HIGH.value, connection=redis_client)
default_queue = Queue(name=QueuePrio.DEFAULT.value, connection=redis_client)
@@ -33,7 +33,7 @@ def __get_sync_cache() -> Redis:
return FakeRedis(version=7)
# A separate client that auto-decodes responses is needed
client = Redis.from_url(str(REDIS_URL), decode_responses=True)
client = Redis.from_url(REDIS_URL, decode_responses=True)
log.debug(
f"Sync redis/valkey connection established in {os.path.splitext(os.path.basename(sys.argv[0]))[0]}"
)
@@ -48,7 +48,7 @@ def __get_async_cache() -> AsyncRedis:
return FakeAsyncRedis(version=7)
# A separate client that auto-decodes responses is needed
client = AsyncRedis.from_url(str(REDIS_URL), decode_responses=True)
client = AsyncRedis.from_url(REDIS_URL, decode_responses=True)
log.debug(
f"Async redis/valkey connection established in {os.path.splitext(os.path.basename(sys.argv[0]))[0]}"
)

2
backend/handler/socket_handler.py
View File

@@ -12,7 +12,7 @@ class SocketHandler:
json=json_module,
logger=False,
engineio_logger=False,
client_manager=socketio.AsyncRedisManager(str(REDIS_URL)),
client_manager=socketio.AsyncRedisManager(REDIS_URL),
ping_timeout=60,
ping_interval=25,
max_http_buffer_size=1e6, # 1MB

8
backend/utils/database.py
View File

@@ -124,6 +124,14 @@ def json_array_contains_all(
)
def safe_str_to_bool(value: Any, default: bool = False) -> bool:
"""Safely convert a value to bool, returning default if conversion fails."""
try:
return value.strip().lower() in ("1", "true", "yes", "on")
except (ValueError, TypeError, AttributeError):
return default
def safe_float(value: Any, default: float = 0.0) -> float:
"""Safely convert a value to float, returning default if conversion fails."""
try: