diff --git a/py3xui/api/__init__.py b/py3xui/api/__init__.py index 5b48bc9..002d902 100644 --- a/py3xui/api/__init__.py +++ b/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 diff --git a/py3xui/api/api.py b/py3xui/api/api.py index 8c2dff3..34efb0e 100644 --- a/py3xui/api/api.py +++ b/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.") diff --git a/py3xui/api/api_base.py b/py3xui/api/api_base.py index 7daa4cf..b465921 100644 --- a/py3xui/api/api_base.py +++ b/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) diff --git a/py3xui/api/api_client.py b/py3xui/api/api_client.py index f101627..3ba7af8 100644 --- a/py3xui/api/api_client.py +++ b/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 [] diff --git a/py3xui/api/api_database.py b/py3xui/api/api_database.py new file mode 100644 index 0000000..51855e5 --- /dev/null +++ b/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.") diff --git a/tests/test_api.py b/tests/test_api.py index 229be65..a78d4e4 100644 --- a/tests/test_api.py +++ b/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()