Add requests to UserSubscription

Author: sm1ky

.gitignore

@@ -71,4 +71,5 @@ venv/
 #openapi
 openapi.json
 temp.py
-bump.sh
+bump.sh
+test.py

example/example.py

@@ -1,5 +1,6 @@
 import asyncio
 from marzban import MarzbanAPI, AdminCreate, UserCreate, NodeCreate, UserTemplateCreate, AdminModify, UserModify, UserTemplateModify, NodeModify, ProxySettings
+from datetime import datetime 
 
 async def main():
     # Create an API client
@@ -77,11 +78,11 @@ async def main():
     print("Set Owner:", owner_set)
 
     ## Get list of expired users
-    expired_users = await api.get_expired_users(token=token.access_token, expired_before="2024-01-01T00:00:00Z")
+    expired_users = await api.get_expired_users(token=token.access_token, expired_before=datetime.fromisoformat("2024-10-10T00:00:00+00:00"))
     print("Expired Users:", expired_users)
 
     ## Delete expired users
-    deleted_expired_users = await api.delete_expired_users(token=token.access_token, expired_before="2024-01-01T00:00:00Z")
+    deleted_expired_users = await api.delete_expired_users(token=token.access_token, expired_before=datetime.fromisoformat("2024-10-10T00:00:00+00:00"))
     print("Deleted Expired Users:", deleted_expired_users)
 
     # User templates management
@@ -136,6 +137,23 @@ async def main():
     nodes_usage = await api.get_usage(token=token.access_token, start="2023-01-01", end="2023-12-31")
     print("Nodes Usage:", nodes_usage)
 
+    ## Getting node settings
+    node_settings = await api.get_node_settings(token=token.access_token)
+    print("Node Settings:", node_settings)
+
+    ## Getting user subscription information with token
+    user_subscription_info = await api.get_user_subscription_info(url=user_info.subscription_url)
+    print("User Subscription Info:", user_subscription_info)
+
+    ## Getting user usage statistics
+    user_usage = await api.get_user_usage(url=user_info.subscription_url, start="2023-01-01")
+    print("User Usage:", user_usage)
+
+    ## Getting user subscription by client type
+    # Accepts only clients that require a JSON response. Example: sing-box.
+    user_subscription_client = await api.get_user_subscription_with_client_type(url=user_info.subscription_url, client_type="sing-box")
+    print("User Subscription with Client Type:", user_subscription_client)
+
     # Closing the API client
     await api.close()
 

marzban/__init__.py

@@ -25,7 +25,9 @@
     ProxyInbound, 
     Token, 
     HTTPValidationError, 
-    ValidationError
+    ValidationError,
+    SubscriptionUserResponse,
+    SystemStats
 )
 
 __all__ = (
@@ -57,6 +59,8 @@
     "Token", 
     "HTTPValidationError", 
     "ValidationError",
+    "SubscriptionUserResponse",
+    "SystemStats"
 )
 
 __version__ = "0.2.8"

marzban/api.py

@@ -223,5 +223,45 @@ async def get_usage(self, token: str, start: Optional[str] = None, end: Optional
         response = await self._request("GET", url, token, params=params)
         return NodesUsageResponse(**response.json())
 
+    async def get_user_subscription_info(self, url: str = None, token: str = None) -> SubscriptionUserResponse:
+        if url:
+            # Use the provided URL if it is given
+            final_url = url + "/info"
+        elif token:
+            # Form the URL using the token if it is provided
+            final_url = f"/sub/{token}/info"
+        else:
+            raise ValueError("Either url or token must be provided")
+        
+        response = await self._request("GET", final_url)
+        return SubscriptionUserResponse(**response.json())
+
+    async def get_user_usage(self, url: str = None, token: str = None, start: Optional[str] = None, end: Optional[str] = None) -> Any:
+        if url:
+            # Use the provided URL if it is given
+            final_url = url + "/usage"
+        elif token:
+            # Form the URL using the token if it is provided
+            final_url = f"/sub/{token}/usage"
+        else:
+            raise ValueError("Either url or token must be provided")
+        
+        params = {"start": start, "end": end}
+        response = await self._request("GET", final_url, params=params)
+        return response.json()
+
+    async def get_user_subscription_with_client_type(self, client_type: str, url: str = None, token: str = None) -> Any:
+        if url:
+            # Use the provided URL if it is given
+            final_url = url + f"/{client_type}"
+        elif token:
+            # Form the URL using the token if it is provided
+            final_url = f"/sub/{token}/{client_type}"
+        else:
+            raise ValueError("Either url or token must be provided")
+        
+        response = await self._request("GET", final_url)
+        return response.json()
+
     async def close(self):
         await self.client.aclose()

marzban/models.py

@@ -1,5 +1,5 @@
-from pydantic import BaseModel
-from typing import Optional, List, Dict, Any, ClassVar
+from pydantic import BaseModel, field_validator, ValidationInfo, AfterValidator, ValidationError
+from typing import Optional, List, Dict, Any, ClassVar, Annotated
 
 class Token(BaseModel):
     access_token: str
@@ -23,19 +23,6 @@ class AdminModify(BaseModel):
 class HTTPValidationError(BaseModel):
     detail: Optional[List[Dict[str, Any]]] = None
 
-class SystemStats(BaseModel):
-    version: str
-    mem_total: int
-    mem_used: int
-    cpu_cores: int
-    cpu_usage: float
-    total_user: int
-    users_active: int
-    incoming_bandwidth: int
-    outgoing_bandwidth: int
-    incoming_bandwidth_speed: int
-    outgoing_bandwidth_speed: int
-
 class ProxySettings(BaseModel):
     id: Optional[str] = None
     flow: Optional[str] = None
@@ -75,7 +62,13 @@ class UserResponse(BaseModel):
     created_at: Optional[str] = None
     links: Optional[List[str]] = []
     subscription_url: Optional[str] = None
+    subscription_token: Optional[str] = None
     excluded_inbounds: Optional[Dict[str, List[str]]] = None
+    
+    def __init__(self, **data):
+        super().__init__(**data)
+        if not self.subscription_token and self.subscription_url:
+            self.subscription_token = self.subscription_url.split('/')[-1]
 
 class NodeCreate(BaseModel):
     name: str
@@ -197,3 +190,39 @@ class ValidationError(BaseModel):
     loc: List[Any]
     msg: str
     type: str
+
+class SubscriptionUserResponse(BaseModel):
+    proxies: Dict[str, Any]
+    expire: Optional[int] = None
+    data_limit: Optional[int] = None
+    data_limit_reset_strategy: str = "no_reset"
+    inbounds: Dict[str, List[str]] = {}
+    note: Optional[str] = None
+    sub_updated_at: Optional[str] = None
+    sub_last_user_agent: Optional[str] = None
+    online_at: Optional[str] = None
+    on_hold_expire_duration: Optional[int] = None
+    on_hold_timeout: Optional[str] = None
+    auto_delete_in_days: Optional[int] = None
+    username: str
+    status: str
+    used_traffic: int
+    lifetime_used_traffic: int = 0
+    created_at: str
+    links: List[str] = []
+    subscription_url: str = ""
+    excluded_inbounds: Dict[str, List[str]] = {}
+    admin: Optional[Admin] = None
+    
+class SystemStats(BaseModel):
+    version: Optional[str] = None
+    mem_total: Optional[int] = None
+    mem_used: Optional[int] = None
+    cpu_cores: Optional[int] = None
+    cpu_usage: Optional[float] = None
+    total_user: Optional[int] = None
+    users_active: Optional[int] = None
+    incoming_bandwidth: Optional[int] = None
+    outgoing_bandwidth: Optional[int] = None
+    incoming_bandwidth_speed: Optional[int] = None
+    outgoing_bandwidth_speed: Optional[int] = None