All methods implemented.

py3xui/api/__init__.py

@@ -1,2 +1,3 @@
 from py3xui.api.api_client import ClientApi
+from py3xui.api.api_database import DatabaseApi
 from py3xui.api.api_inbound import InboundApi

py3xui/api/api.py

@@ -1,6 +1,6 @@
 """This module provides classes to interact with the XUI API."""
 
-from py3xui.api import ClientApi, InboundApi
+from py3xui.api import ClientApi, DatabaseApi, InboundApi
 from py3xui.utils import Logger, env
 
 logger = Logger(__name__)
@@ -10,6 +10,7 @@ class Api:
     def __init__(self, host: str, username: str, password: str, skip_login: bool = False):
         self.client = ClientApi(host, username, password)
         self.inbound = InboundApi(host, username, password)
+        self.database = DatabaseApi(host, username, password)
         if not skip_login:
             self.login()
 
@@ -23,4 +24,5 @@ def from_env(cls, skip_login: bool = False):
     def login(self) -> None:
         self.client.login()
         self.inbound.session = self.client.session
+        self.database.session = self.client.session
         logger.info("Logged in successfully.")

py3xui/api/api_base.py

@@ -91,8 +91,11 @@ def _request_with_retry(
         logger.debug("%s request to %s...", method.__name__.upper(), url)
         for retry in range(1, self.max_retries + 1):
             try:
+                skip_check = kwargs.pop("skip_check", False)
                 response = method(url, cookies={"session": self.session}, headers=headers, **kwargs)
                 response.raise_for_status()
+                if skip_check:
+                    return response
                 self._check_response(response)
                 return response
             except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e:
@@ -108,8 +111,10 @@ def _request_with_retry(
             f"Max retries exceeded with no successful response to {url}"
         )
 
-    def _post(self, url: str, headers: dict[str, str], data: dict[str, Any]) -> requests.Response:
-        return self._request_with_retry(requests.post, url, headers, json=data)
+    def _post(
+        self, url: str, headers: dict[str, str], data: dict[str, Any], **kwargs
+    ) -> requests.Response:
+        return self._request_with_retry(requests.post, url, headers, json=data, **kwargs)
 
-    def _get(self, url: str, headers: dict[str, str]) -> requests.Response:
-        return self._request_with_retry(requests.get, url, headers)
+    def _get(self, url: str, headers: dict[str, str], **kwargs) -> requests.Response:
+        return self._request_with_retry(requests.get, url, headers, **kwargs)

py3xui/api/api_client.py

@@ -133,3 +133,26 @@ def delete(self, inbound_id: int, client_uuid: str) -> None:
 
         self._post(url, headers, data)
         logger.info("Client deleted successfully.")
+
+    def delete_depleted(self, inbound_id: int) -> None:
+        endpoint = f"panel/api/inbounds/delDepletedClients/{inbound_id}"
+        headers = {"Accept": "application/json"}
+
+        url = self._url(endpoint)
+        data: dict[str, Any] = {}
+        logger.info("Deleting depleted clients for inbound ID: %s", inbound_id)
+
+        self._post(url, headers, data)
+        logger.info("Depleted clients deleted successfully.")
+
+    def online(self) -> list[str]:
+        endpoint = "panel/api/inbounds/onlines"
+        headers = {"Accept": "application/json"}
+
+        url = self._url(endpoint)
+        data: dict[str, Any] = {}
+        logger.info("Getting online clients")
+
+        response = self._post(url, headers, data)
+        online = response.json().get(ApiFields.OBJ)
+        return online or []

py3xui/api/api_database.py

@@ -0,0 +1,16 @@
+from py3xui.api.api_base import BaseApi
+from py3xui.utils import Logger
+
+logger = Logger(__name__)
+
+
+class DatabaseApi(BaseApi):
+    def export(self) -> None:
+        endpoint = "panel/api/inbounds/createbackup"
+        headers = {"Accept": "application/json"}
+
+        url = self._url(endpoint)
+        logger.info("Exporting database...")
+
+        self._get(url, headers, skip_check=True)
+        logger.info("Database exported successfully.")

tests/test_api.py

@@ -195,3 +195,31 @@ def test_reset_client_stats():
         )
         api = Api(HOST, USERNAME, PASSWORD, skip_login=True)
         api.client.reset_stats(1, EMAIL)
+
+
+def test_delete_client():
+    with requests_mock.Mocker() as m:
+        m.post(f"{HOST}/panel/api/inbounds/1/delClient/1", json={ApiFields.SUCCESS: True})
+        api = Api(HOST, USERNAME, PASSWORD, skip_login=True)
+        api.client.delete(1, "1")
+
+
+def test_delete_depleted_clients():
+    with requests_mock.Mocker() as m:
+        m.post(f"{HOST}/panel/api/inbounds/delDepletedClients/1", json={ApiFields.SUCCESS: True})
+        api = Api(HOST, USERNAME, PASSWORD, skip_login=True)
+        api.client.delete_depleted(1)
+
+
+def test_client_online():
+    with requests_mock.Mocker() as m:
+        m.post(f"{HOST}/panel/api/inbounds/onlines", json={ApiFields.SUCCESS: True})
+        api = Api(HOST, USERNAME, PASSWORD, skip_login=True)
+        api.client.online()
+
+
+def test_database_export():
+    with requests_mock.Mocker() as m:
+        m.get(f"{HOST}/panel/api/inbounds/createbackup", json={ApiFields.SUCCESS: True})
+        api = Api(HOST, USERNAME, PASSWORD, skip_login=True)
+        api.database.export()