From fbd7c19a93e64225f909c336396fe9cb3f20e13f Mon Sep 17 00:00:00 2001 From: Simon Date: Sun, 22 Dec 2024 22:59:48 +0700 Subject: [PATCH] remove dependency on redis json --- backend/appsettings/src/config.py | 5 +--- backend/appsettings/views.py | 2 +- backend/common/src/ta_redis.py | 30 ++++++++++++++----- .../config/management/commands/ta_startup.py | 10 ++++++- backend/download/src/yt_dlp_base.py | 10 +++---- backend/task/src/task_manager.py | 14 ++------- backend/task/views.py | 7 +---- backend/video/views.py | 2 +- 8 files changed, 43 insertions(+), 37 deletions(-) diff --git a/backend/appsettings/src/config.py b/backend/appsettings/src/config.py index c14626e6..24c3c13f 100644 --- a/backend/appsettings/src/config.py +++ b/backend/appsettings/src/config.py @@ -238,8 +238,5 @@ class ReleaseVersion: def get_update(self) -> dict: """return new version dict if available""" - message = RedisArchivist().get_message(self.NEW_KEY) - if not message.get("status"): - return {} - + message = RedisArchivist().get_message_dict(self.NEW_KEY) return message diff --git a/backend/appsettings/views.py b/backend/appsettings/views.py index cf224e88..26c47acc 100644 --- a/backend/appsettings/views.py +++ b/backend/appsettings/views.py @@ -198,7 +198,7 @@ class CookieView(ApiBaseView): """handle get request""" # pylint: disable=unused-argument config = AppConfig().config - valid = RedisArchivist().get_message("cookie:valid") + valid = RedisArchivist().get_message_dict("cookie:valid") response = {"cookie_enabled": config["downloads"]["cookie_import"]} response.update(valid) diff --git a/backend/common/src/ta_redis.py b/backend/common/src/ta_redis.py index fbb3a628..4390c326 100644 --- a/backend/common/src/ta_redis.py +++ b/backend/common/src/ta_redis.py @@ -40,15 +40,15 @@ class RedisArchivist(RedisBase): def set_message( self, key: str, - message: dict, - path: str = ".", + message: dict | str, expire: bool | int = False, save: bool = False, ) -> None: """write new message to redis""" - self.conn.execute_command( - "JSON.SET", self.NAME_SPACE + key, path, json.dumps(message) + to_write = ( + json.dumps(message) if isinstance(message, dict) else message ) + self.conn.execute_command("SET", self.NAME_SPACE + key, to_write) if expire: if isinstance(expire, bool): @@ -67,8 +67,24 @@ class RedisArchivist(RedisBase): except redis.exceptions.ResponseError: pass - def get_message(self, key: str) -> dict: - """get message dict from redis""" + def get_message_str(self, key: str) -> str | None: + """get message string""" + reply = self.conn.execute_command("GET", self.NAME_SPACE + key) + return reply + + def get_message_dict(self, key: str) -> dict: + """get message dict""" + reply = self.conn.execute_command("GET", self.NAME_SPACE + key) + if not reply: + return {} + + return json.loads(reply) + + def get_message(self, key: str) -> dict | None: + """ + get message dict from redis + old json get message, only used for migration, to be removed later + """ reply = self.conn.execute_command("JSON.GET", self.NAME_SPACE + key) if reply: return json.loads(reply) @@ -91,7 +107,7 @@ class RedisArchivist(RedisBase): if not all_matches: return [] - return [self.get_message(i) for i in all_matches] + return [self.get_message_dict(i) for i in all_matches] def del_message(self, key: str) -> bool: """delete key from redis""" diff --git a/backend/config/management/commands/ta_startup.py b/backend/config/management/commands/ta_startup.py index 2ea6597f..c995f5c0 100644 --- a/backend/config/management/commands/ta_startup.py +++ b/backend/config/management/commands/ta_startup.py @@ -19,6 +19,7 @@ from common.src.ta_redis import RedisArchivist from django.core.management.base import BaseCommand, CommandError from django.utils import dateformat from django_celery_beat.models import CrontabSchedule, PeriodicTasks +from redis.exceptions import ResponseError from task.models import CustomPeriodicTask from task.src.config_schedule import ScheduleBuilder from task.src.task_manager import TaskManager @@ -54,7 +55,14 @@ class Command(BaseCommand): def _mig_app_settings(self) -> None: """update from v0.4.10 to v0.5.0, migrate application settings""" self.stdout.write("[MIGRATION] move appconfig to ES") - config = RedisArchivist().get_message("config") + try: + config = RedisArchivist().get_message("config") + except ResponseError: + self.stdout.write( + self.style.SUCCESS(" Redis does not support JSON decoding") + ) + return + if not config: self.stdout.write( self.style.SUCCESS(" no config values to migrate") diff --git a/backend/download/src/yt_dlp_base.py b/backend/download/src/yt_dlp_base.py index 08013ff4..31a7e73a 100644 --- a/backend/download/src/yt_dlp_base.py +++ b/backend/download/src/yt_dlp_base.py @@ -10,6 +10,7 @@ from http import cookiejar from io import StringIO import yt_dlp +from appsettings.src import config as ConfigApp from common.src.env_settings import EnvironmentSettings from common.src.ta_redis import RedisArchivist @@ -91,7 +92,7 @@ class CookieHandler: def get(self): """get cookie io stream""" - cookie = RedisArchivist().get_message("cookie") + cookie = RedisArchivist().get_message_str("cookie") self.cookie_io = StringIO(cookie) return self.cookie_io @@ -116,8 +117,7 @@ class CookieHandler: def set_cookie(self, cookie): """set cookie str and activate in config""" RedisArchivist().set_message("cookie", cookie, save=True) - path = ".downloads.cookie_import" - RedisArchivist().set_message("config", True, path=path, save=True) + ConfigApp.AppConfig().update_config({"downloads.cookie_import": True}) self.config["downloads"]["cookie_import"] = True print("cookie: activated and stored in Redis") @@ -126,9 +126,7 @@ class CookieHandler: """revoke cookie""" RedisArchivist().del_message("cookie") RedisArchivist().del_message("cookie:valid") - RedisArchivist().set_message( - "config", False, path=".downloads.cookie_import" - ) + ConfigApp.AppConfig().update_config({"downloads.cookie_import": False}) print("cookie: revoked") def validate(self): diff --git a/backend/task/src/task_manager.py b/backend/task/src/task_manager.py index 8606d54b..ea48e5e8 100644 --- a/backend/task/src/task_manager.py +++ b/backend/task/src/task_manager.py @@ -4,9 +4,8 @@ functionality: - handle threads and locks """ -from common.src.ta_redis import RedisArchivist, TaskRedis +from common.src.ta_redis import TaskRedis from task.celery import app as celery_app -from task.src.task_config import TASK_CONFIG class TaskManager: @@ -96,20 +95,13 @@ class TaskCommand: return message - def stop(self, task_id, message_key): + def stop(self, task_id): """ send stop signal to task_id, needs to be implemented in task to take effect """ print(f"[task][{task_id}]: received STOP signal.") - handler = TaskRedis() - - task = handler.get_single(task_id) - if not task["name"] in TASK_CONFIG: - raise ValueError - - handler.set_command(task_id, "STOP") - RedisArchivist().set_message(message_key, "STOP", path=".command") + TaskRedis().set_command(task_id, "STOP") def kill(self, task_id): """send kill signal to task_id""" diff --git a/backend/task/views.py b/backend/task/views.py index 7965fb38..61d03f7f 100644 --- a/backend/task/views.py +++ b/backend/task/views.py @@ -103,8 +103,7 @@ class TaskIDView(ApiBaseView): message = {"message": "task can not be stopped"} return Response(message, status=400) - message_key = self._build_message_key(task_conf, task_id) - TaskCommand().stop(task_id, message_key) + TaskCommand().stop(task_id) if command == "kill": if not task_conf.get("api_stop"): message = {"message": "task can not be killed"} @@ -114,10 +113,6 @@ class TaskIDView(ApiBaseView): return Response({"message": "command sent"}) - def _build_message_key(self, task_conf, task_id): - """build message key to forward command to notification""" - return f"message:{task_conf.get('group')}:{task_id.split('-')[0]}" - class ScheduleListView(ApiBaseView): """resolves to /api/task/schedule/ diff --git a/backend/video/views.py b/backend/video/views.py index e3ae50d1..294236f0 100644 --- a/backend/video/views.py +++ b/backend/video/views.py @@ -102,7 +102,7 @@ class VideoProgressView(ApiBaseView): """get progress for a single video""" user_id = request.user.id key = f"{user_id}:progress:{video_id}" - video_progress = RedisArchivist().get_message(key) + video_progress = RedisArchivist().get_message_dict(key) position = video_progress.get("position", 0) self.response = {