✏️ Editing: redis_lib.py Read Only

# -*- coding: utf-8 -*-

# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT

from __future__ import absolute_import

import subprocess
from typing import Optional, Tuple, Dict
from clwpos import gettext as _

from clcommon.utils import ExternalProgramFailed, run_command
from clwpos.cl_wpos_exceptions import WposError

class DeadRedisPurgeAttempt(WposError):
    """
    Happens when somebody tries to purge
    redis instance which is not runnning
    """

def __init__(self):
        super().__init__(message=_(
            "Unable to purge cache because cache database is not running. "
            "Enable at least one optimization feature to start cache database instance."))

class PurgeFailedError(WposError):
    """
    Happens when redis is not able to purge cached data for some reason.
    """

def __init__(self, std_out: str, std_err: str):
        super().__init__(
            message=_('Unable to purge cache. \n'
                      'Database backend returned error of command "%(command)s" execution. '
                      'Try again a bit later or contact your system administrator for help.'),
            details='stdout: \n<i>%(stdout)s</i>\n'
                    'stderr: \n<i>%(stderr)s</i>\n',

context={
                "command": 'purge',
                "stdout": std_out or 'empty',
                "stderr": std_err or 'empty',
            })

class RedisLibUser(object):

def __init__(self, socket_path: str,
                 redis_cli_path: str = "/opt/alt/redis/bin/redis-cli") -> None:
        self.socket_path = socket_path
        self.redis_cli_path = redis_cli_path

def get_redis_shell_command(self, redis_command):
        return [
            self.redis_cli_path,
            "-s",
            self.socket_path,
        ] + redis_command.split(" ")

def run_redis_command(self, command: str) -> Tuple[int, str, str]:
        redis_command = self.get_redis_shell_command(command)
        try:
            returncode, std_out, std_err = run_command(
                redis_command, return_full_output=True
            )
        except (subprocess.CalledProcessError, ExternalProgramFailed) as error:
            raise WposError(
                message=_("Error during %(command)s command execution: \n%(error)s"),
                context={"command": " ".join(redis_command), "error": str(error)},
            )
        return returncode, std_out, std_err

def purge_redis(self) -> Dict:
        """
        Clean entire redis cache for user.
        """
        if not self.is_redis_alive():
            raise DeadRedisPurgeAttempt()

purge_command = "flushall async"
        returncode, std_out, std_err = self.run_redis_command(purge_command)
        # Output of Redis FLUSHALL command should be: "OK"
        if returncode != 0 or "ok" not in std_out.lower():
            raise PurgeFailedError(std_out, std_err)

return {
            "used_memory": self.get_redis_used_memory()
        }

def is_redis_alive(self) -> bool:
        """
        Check if user's redis is alive.
        """
        returncode, std_out, _ = self.run_redis_command("ping")
        return returncode == 0 and "pong" in std_out.lower()

def is_redis_empty(self) -> bool:
        """
        Check if user's redis is empty.
        Example (redis is empty):
            # Keyspace
        Example (redis is NOT empty):
            # Keyspace
            db0:keys=2,expires=0,avg_ttl=0
        """
        returncode, std_out, _ = self.run_redis_command("info keyspace")
        return returncode == 0 and len(std_out.strip().split("\n")) <= 1

def get_redis_used_memory(self) -> Optional[str]:
        """
        Return amount of memmory used by user's redis instance
        in human readable format (kb or mb).
        If redis status is offline, 'used_memory_value' value is null.
        'info memory' command output:
            # Memory
            used_memory:369616
            used_memory_human:360.95K
            ...
        """
        _, std_out, _ = self.run_redis_command("info memory")
        for line in std_out.split("\n"):
            if not line.startswith("used_memory_human"):
                continue
            _, used_memory_value = line.split(":")
            return used_memory_value if used_memory_value != "null" else None

🔐 Sid Gifari File Manager Pro