✏️ Editing: user_limits.py Read Only

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

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

"""
This module provides functions to retrieve any limits and usage for a specified user.
"""
import subprocess
import json

class LveCommandError(Exception):
    def __str__(self):
        return f"{self.args[0]}"

def get_lve_limits(username):
    """
    Retrieves the LVE limits for a specified user.

Args:
        username (str): The username to retrieve LVE limits for.

Returns:
        When user limits are found
        {
            "lve_cpu": {
                "limit": int
            },
            "lve_ep": {
                "limit": int
            },
            "lve_pmem": {
                "limit": int
            },
            "lve_iops": {
                "limit": int
            },
            "lve_io": {
                "limit": int
            },
            "lve_nproc": {
                "limit": int
            },
        }
        or an error
        {
            "error": str
        }
    """
    cmd = ["/usr/sbin/cloudlinux-limits", "get", "--username", username, "--json"]
    try:
        stdout = _call_lve_command(cmd)
        limits = json.loads(stdout)
    except LveCommandError as e:
        return {"error": str(e)}
    except json.JSONDecodeError as e:
        return {"error": f"Failed to decode JSON: {e}"}

if 'users' not in limits or not any(user['username'] == username for user in limits['users']):
        return {"error": f"LVE Limit for user {username} not found"}

user_limits = next(user['limits'] for user in limits['users'] if user['username'] == username)

return {
        "lve_cpu": {
            "limit": int(user_limits['cpu']['all'].strip('*'))  # %
        },
        "lve_ep": {
            "limit": int(user_limits['ep'].strip('*'))  # proc num
        },
        "lve_pmem": {
            "limit": int(user_limits['pmem'].strip('*'))  # byte
        },
        "lve_iops": {
            "limit": int(user_limits['iops'].strip('*'))  # op per sec
        },
        "lve_io": {
            "limit": int(user_limits['io']['all'].strip('*')) * 1024  # byte
        },
        "lve_nproc": {
            "limit": int(user_limits['nproc'].strip('*'))  # proc num
        },
    }

def get_lve_usage(username):
    """
    Retrieves the current LVE usage for a specified user.

Args:
        username (str): The username to retrieve LVE usage for.

Returns:
        When user usage are found
        {
            "lve_cpu": {
                "usage": int
            },
            "lve_ep": {
                "usage": int
            },
            "lve_pmem": {
                "usage": int
            },
            "lve_iops": {
                "usage": int
            },
            "lve_io": {
                "usage": int
            },
            "lve_nproc": {
                "usage": int
            },
        }
        or empty dict when user sleep
        {}
        or an error
        {
            "error": str
        }
    """
    cmd = ["/usr/sbin/lveps", "-d", "-o", "id:15,ep:7,pno:7,cpu:7,mem:9,io:11,iops:11"]
    try:
        stdout = _call_lve_command(cmd)
        lines = stdout.strip().split('\n')
    except LveCommandError as e:
        return {"error": str(e)}

usage = {}
    for line in lines[1:]:
        fields = line.split()
        if fields[0] == username:
            usage = {
                "lve_cpu": {
                    "usage": int(fields[3].strip('%'))  # (SPEED/CPU) %
                },
                "lve_ep": {
                    "usage": int(fields[1])  # (EP) proc num
                },
                "lve_pmem": {
                    "usage": int(fields[4]) * 1024 * 1024  # (MEM) byte
                },
                "lve_iops": {
                    "usage": int(fields[6])  # (IOPS) op per sec
                },
                "lve_io": {
                    "usage": int(fields[5])  # (IO) byte
                },
                "lve_nproc": {
                    "usage": int(fields[2])  # (PNO) proc num
                },
            }
            break

return usage

def _call_lve_command(cmd):
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, check=True)
        return result.stdout
    except subprocess.CalledProcessError as e:
        raise LveCommandError(f"Command '{cmd}' failed with error: {e}")
    except Exception as e:
        raise LveCommandError(f"An unexpected error occurred: {e}")

🔐 Sid Gifari File Manager Pro