A Python library that helps you easily use Marzban's API panel
from .marzban import Marzban

marzpy/__init__.py
mewhrzad/marzpy
from .admin import Admin
from .node import NodeMethods
from .subscription import Subscription
from .core import Core
from .user import UserMethods
from .template import TemplateMethods
from .system import System


class Methods(
    Admin, NodeMethods, Subscription, Core, UserMethods, TemplateMethods, System
):
    def __init__(self, username: str, password: str, panel_address: str):
        super().__init__(username, password, panel_address)

marzpy/api/__init__.py
mewhrzad/marzpy
from .send_requests import *


class Admin:
    def __init__(self, username: str, password: str, panel_address: str):
        self.username = username
        self.password = password
        self.panel_address = panel_address

    def get_token(self):
        """login for Authorization token

        Returns: `~dict`: Authorization token
        """
        try:
            request_address = f"{self.panel_address}/api/admin/token"
            payload = {"username": self.username, "password": self.password}
            response = requests.post(
                request_address,
                data=payload,
                timeout=3000,
                headers={
                    "Accept": "application/json",
                    "Content-Type": "application/x-www-form-urlencoded",
                },
            )
            response.raise_for_status()  # Raise an exception for non-200 status codes
            result = json.loads(response.content)
            result["panel_address"] = self.panel_address
            return result
        except requests.exceptions.RequestException as ex:
            print(f"Request Exception: {ex}")
            return None
        except json.JSONDecodeError as ex:
            print(f"JSON Decode Error: {ex}")
            return None

    def get_current_admin(self, token: dict):
        """get current admin who has logged in.

        Parameters:
            token (``dict``) : Authorization token

        Returns:
        `~dict`: {"username": "str" , "is_sudo": true}
        """
        return send_request(endpoint="admin", token=token, method="get")

    def create_admin(self, token: dict, data: dict):
        """add new admin.

        Parameters:
            token (``dict``) : Authorization token
            data (``dict``) : information of new admin

        Returns:
        `~dict`: username && is_sudo
        """
        send_request(endpoint="admin", token=token, method="post", data=data)
        return "success"

    def change_admin_password(self, username: str, token: dict, data: dict):
        """change exist admins password.

        *you cant modify sudo admins password*

        Parameters:
            username (``str``) : username of admin
            token (``dict``) : Authorization token
            data (``dict``) : information of new admin

        Returns:
        `~dict`: username && is_sudo
        """
        send_request(
            endpoint=f"admin/{username}",
            token=token,
            method="put",
            data=data,
        )
        return "success"

    def delete_admin(self, username: str, token: dict):
        """delete admin.

        Parameters:
            username (``str``) : username of admin
            token (``dict``) : Authorization token

        Returns:
        `~str`: success
        """
        send_request(endpoint=f"admin/{username}", token=token, method="delete")
        return "success"

    def get_all_admins(self, token: dict):
        """get all admins.

        Parameters:
            token (``dict``) : Authorization token

        Returns:
        `~list`: [{username && is_sudo}]
        """
        return send_request(endpoint=f"admins", token=token, method="get")

marzpy/api/admin.py
mewhrzad/marzpy
from .send_requests import *


class Core:
    def __init__(self) -> None:
        pass

    def get_xray_core(self, token: dict):
        """get xray core.

        Parameters:
            token (``dict``): Authorization token

        Returns:
            `~dict`: xray core
        """
        return send_request(endpoint="core", token=token, method="get")

    def restart_xray_core(self, token: dict):
        """restart xray core.

        Parameters:
            token (``dict``): Authorization token

        Returns:
            `~str`: success
        """
        send_request(endpoint="core/restart", token=token, method="post")
        return "success"

    def get_xray_config(self, token: dict):
        """get xray config.

        Parameters:
            token (``dict``): Authorization token

        Returns:
            `~dict`: xray config
        """
        return send_request(endpoint="core/config", token=token, method="get")

    def modify_xray_config(self, token: dict, config: json):
        """edit xray config.

        Parameters:
            token (``dict``): Authorization token
            config (``json``): json of new config

        Returns:
            `~str`: success
        """
        send_request(endpoint="core/config", token=token, method="put", data=config)
        return "success"

marzpy/api/core.py
mewhrzad/marzpy
from .send_requests import *


class Node:
    def __init__(
        self,
        name="",
        address="",
        port=0,
        api_port=0,
        certificate="",
        id=0,
        xray_version="",
        status="",
        message="",
    ):
        self.name = name
        self.address = address
        self.port = port
        self.api_port = api_port
        self.certificate = certificate
        self.id = id
        self.xray_version = xray_version
        self.status = status
        self.message = message


class NodeMethods:
    def __init__(self) -> None:
        pass

    def add_node(self, token: dict, node: Node):
        """add new node.

        Parameters:
            token (``dict``): Authorization token

            node (``api.Node``): node object

        Returns:
            `~object`: information of new node
        """
        return Node(
            **send_request(
                endpoint="node", token=token, method="post", data=node.__dict__
            )
        )

    def get_node_by_id(self, id: int, token: dict):
        """get exist node from id.

        Parameters:
            id (``int``): id of node

            token (``dict``): Authorization token

        Returns:
            `~object`: information of new node
        """
        return Node(**send_request(endpoint=f"node/{id}", token=token, method="get"))

    def modify_node_by_id(self, id: int, token: dict, node: object):
        """edit exist node from id.

        Parameters:
            id (``int``): id of node

            token (``dict``): Authorization token

            node (``api.Node``): node object

        Returns:
            `~object`: information of new node
        """
        request = send_request(
            endpoint=f"node/{id}", token=token, method="put", data=node.__dict__
        )
        return Node(**request)

    def delete_node(self, id: int, token: dict):
        """delete node from id.

        Parameters:
            id (``int``): id of node

            token (``dict``): Authorization token

        Returns:
            `~str`: success
        """
        send_request(endpoint=f"node/{id}", token=token, method="delete")
        return "success"

    def get_all_nodes(self, token: dict):
        """get all nodes.

        Parameters:
            token (``dict``): Authorization token

        Returns:
            `~list of objects`: [Node]
        """
        request = send_request(endpoint="nodes", token=token, method="get")
        node_list = [Node()]
        for node in request:
            node_list.append(Node(**node))
        del node_list[0]
        return node_list

    def reconnect_node(self, id: int, token: dict):
        """reconnect from id.

        Parameters:
            id (``int``): id of node

            token (``dict``): Authorization token

        Returns:
            `~str`: success
        """
        request = send_request(
            endpoint=f"node/{id}/reconnect", token=token, method="post"
        )

        return "success"

    def get_nodes_usage(self, token: dict):
        """get all nodes usage.

        Parameters:
            token (``dict``): Authorization token

        Returns:
            `~dict`: "usage" : []
        """
        request = send_request(endpoint="nodes/usage", token=token, method="get")
        return request["usages"]

marzpy/api/node.py
mewhrzad/marzpy
import requests, json


def send_request(endpoint, token, method, data=None):
    try:
        panel_address = token["panel_address"]
        token_type = token["token_type"]
        access_token = token["access_token"]
        request_address = f"{panel_address}/api/{endpoint}"
        headers = {
            "accept": "application/json",
            "Authorization": f"{token_type} {access_token}",
        }
        response = requests.request(
            method, request_address, headers=headers, data=json.dumps(data)
        )
        # print(response.content)
        response.raise_for_status()  # Raise an exception for non-200 status codes
        result = json.loads(response.content)
        return result
    except requests.exceptions.RequestException as ex:
        if response.content:
            raise Exception(f"Request Exception: { response.content }")
        else:
            raise ex
    except json.JSONDecodeError as ex:
        raise f"JSON Decode Error: {ex}"

marzpy/api/send_requests.py
mewhrzad/marzpy
import requests, json, base64


class Subscription:
    def __init__(self) -> None:
        pass

    def subsend_request(sub_link: str, endpoint: str):
        try:
            request_address = f"{sub_link}/{endpoint}"
            headers = {
                "Accept": "application/json",
            }
            response = requests.request("get", request_address, headers=headers)
            response.raise_for_status()  # Raise an exception for non-200 status codes
            result = response.content
            if endpoint:
                return json.loads(response.content)
            else:
                return base64.b64decode(result).decode("utf-8")
        except requests.exceptions.RequestException as ex:
            print(f"Request Exception: {ex}")
            return None

    def get_subscription(self, sub_link: str):
        """Unknow usage!"""
        return Subscription.subsend_request(sub_link, "")

    def get_subscription_info(self, sub_link: str):
        """get user information.

        Parameters:
            token (``dict``): subscription token

        Returns:
            `~dict`: information of user
        """
        return Subscription.subsend_request(sub_link, "info")

marzpy/api/subscription.py
mewhrzad/marzpy
from .send_requests import send_request


class System:
    def __init__(self) -> None:
        pass

    def get_system_stats(self, token: dict):
        """get server stats.

        Parameters:
            token (``dict``): Authorization token

        Returns:
            `~dict`: server stats
        """
        return send_request(endpoint="system", token=token, method="get")

    def get_inbounds(self, token: dict):
        """get server inbounds.

        Parameters:
            token (``dict``): Authorization token

        Returns:
            `~dict`: server inbounds
        """
        return send_request(endpoint="inbounds", token=token, method="get")

    def get_hosts(self, token: dict):
        """get server hosts.

        Parameters:
            token (``dict``): Authorization token

        Returns:
            `~dict`: server hosts
        """
        return send_request(endpoint="hosts", token=token, method="get")

    def modify_hosts(self, token: dict, data: dict):
        """get server hosts.

        Parameters:
            token (``dict``): Authorization token
            data (``dict``) : new hosts data
        Returns:
            `~dict`: server hosts
        """
        return send_request(endpoint="hosts", token=token, method="put", data=data)

marzpy/api/system.py
mewhrzad/marzpy
from .send_requests import *


class Template:
    def __init__(
        self,
        name="",
        inbounds={},
        data_limit={},
        expire_duration=0,
        username_prefix="",
        username_suffix="",
        id=None,
    ):
        self.name = name
        self.inbounds = inbounds
        self.data_limit = data_limit
        self.expire_duration = expire_duration
        self.username_prefix = username_prefix
        self.username_suffix = username_suffix
        self.id = id


class TemplateMethods:
    def get_all_templates(self, token: dict):
        """get all templates list.

        Parameters:
            token (``dict``) : Authorization token

        Returns:
            `~list`: list of templates
        """
        request = send_request(endpoint="user_template", token=token, method="get")
        template_list = [Template()]
        for user in request:
            template_list.append(Template(**user))
        del template_list[0]
        return template_list

    def add_template(self, template: Template, token: dict):
        """add new template.

        Parameters:
            token (``dict``) : Authorization token
            template (``api.template object``) : template

        Returns:
            `~object`: information of new template
        """
        request = send_request(
            endpoint="user_template", token=token, method="post", data=template.__dict__
        )
        return Template(**request)

    def get_template_by_id(self, id: int, token: dict):
        """get exist template from id.

        Parameters:
            id (``id``) : template id
            token (``dict``) : Authorization token
        Returns:
            `~object`: information of template
        """
        request = send_request(
            endpoint=f"user_template/{id}", token=token, method="get"
        )

        return Template(**request)

    def modify_template_by_id(self, id: int, token: dict, template: Template):
        """edit exist template from id.

        Parameters:
            id (``id``) : template id
            token (``dict``) : Authorization token
            template (``object``) template
        Returns:
            `~object`: information of edited template
        """
        request = send_request(
            endpoint=f"user_template/{id}",
            token=token,
            method="put",
            data=template.__dict__,
        )
        return Template(**request)

    def delete_template_by_id(self, id: int, token: dict):
        """delete template from id.

        Parameters:
            id (``id``) : template id
            token (``dict``) : Authorization token
        Returns:
            `~str`: success
        """
        send_request(endpoint=f"user_template/{id}", token=token, method="delete")
        return "success"

marzpy/api/template.py
mewhrzad/marzpy
from .send_requests import *


class User:
    def __init__(
        self,
        username: str,
        proxies: dict,
        inbounds: dict,
        expire: float,
        data_limit: float,
        data_limit_reset_strategy: str,
        status="",
        used_traffic=0,
        lifetime_used_traffic=0,
        created_at="",
        links=[],
        subscription_url="",
        excluded_inbounds={},
    ):
        self.username = username
        self.proxies = proxies
        self.inbounds = inbounds
        self.expire = expire
        self.data_limit = data_limit
        self.data_limit_reset_strategy = data_limit_reset_strategy
        self.status = status
        self.used_traffic = used_traffic
        self.lifetime_used_traffic = lifetime_used_traffic
        self.created_at = created_at
        self.links = links
        self.subscription_url = subscription_url
        self.excluded_inbounds = excluded_inbounds


class UserMethods:
    def add_user(self, user: User, token: dict):
        """add new user.

        Parameters:
            user (``api.User``) : User Object

            token (``dict``) : Authorization token

        Returns: `~User`: api.User object
        """
        request = send_request(
            endpoint="user", token=token, method="post", data=user.__dict__
        )

        return User(**request)

    def get_user(self, user_username: str, token: dict):
        """get exist user information by username.

        Parameters:
            user_username (``str``) : username of user

            token (``dict``) : Authorization token

        Returns: `~User`: api.User object
        """
        request = send_request(f"user/{user_username}", token=token, method="get")
        return User(**request)

    def modify_user(self, user_username: str, token: dict, user: object):
        """edit exist user by username.

        Parameters:
            user_username (``str``) : username of user

            token (``dict``) : Authorization token

            user (``api.User``) : User Object

        Returns: `~User`: api.User object
        """
        request = send_request(f"user/{user_username}", token, "put", user.__dict__)
        return User(**request)

    def delete_user(self, user_username: str, token: dict):
        """delete exist user by username.

        Parameters:
            user_username (``str``) : username of user

            token (``dict``) : Authorization token

        Returns: `~str`: success
        """
        send_request(f"user/{user_username}", token, "delete")
        return "success"

    def reset_user_traffic(self, user_username: str, token: dict):
        """reset exist user traffic by username.

        Parameters:
            user_username (``str``) : username of user

            token (``dict``) : Authorization token

        Returns: `~str`: success
        """
        send_request(f"user/{user_username}/reset", token, "post")
        return "success"

    def get_all_users(self, token: dict):
        """get all users list.

        Parameters:
            token (``dict``) : Authorization token

        Returns:
            `~list`: list of users
        """
        request = send_request("users", token, "get")
        user_list = [
            User(
                username="",
                proxies={},
                inbounds={},
                expire=0,
                data_limit=0,
                data_limit_reset_strategy="",
            )
        ]
        for user in request["users"]:
            user_list.append(User(**user))
        del user_list[0]
        return user_list

    def reset_all_users_traffic(self, token: dict):
        """reset all users traffic.

        Parameters:
            token (``dict``) : Authorization token

        Returns: `~str`: success
        """
        send_request("users/reset", token, "post")
        return "success"

    def get_user_usage(self, user_username: str, token: dict):
        """get user usage by username.

        Parameters:
            user_username (``str``) : username of user

            token (``dict``) : Authorization token

        Returns: `~dict`: dict of user usage
        """
        return send_request(f"user/{user_username}/usage", token, "get")["usages"]

    def get_all_users_count(self, token: dict):
        """get all users count.

        Parameters:
            token (``dict``) : Authorization token

        Returns: `~int`: count of users
        """
        return self.get_all_users(token)["content"]["total"]

marzpy/api/user.py
mewhrzad/marzpy
from .api import Methods


class Marzban(Methods):
    def __init__(self, username: str, password: str, panel_address: str) -> None:
        super().__init__(username, password, panel_address)
        self.username = username
        self.password = password
        self.panel_address = panel_address

marzpy/marzban.py
mewhrzad/marzpy
from setuptools import setup, find_packages

with open("README.md", "r") as file:
    readme = file.read()

setup(
    name="marzpy",
    version="0.0.1",
    author="Mewhrzad",
    description="a simple application with python to manage Marzban panel",
    long_description="text/markdown",
    url="https://github.com/Mewhrzad/marzpy",
    keywords=["marzpy", "Marzban", "Gozargah", "Marzban python", "Marzban API"],
    packages=find_packages(),
    ins=["requests"],
    classifiers=["Programming Language :: Python :: 3"],
)

setup.py
mewhrzad/marzpy