Implementation of automatic token renewal with a usage example

Thank you for the improvements!

Author: sm1ky

example/example.py

@@ -2,7 +2,7 @@
 from datetime import datetime
 
 from marzban import MarzbanAPI, AdminCreate, UserCreate, NodeCreate, UserTemplateCreate, AdminModify, UserModify, \
-    UserTemplateModify, NodeModify, ProxySettings
+    UserTemplateModify, NodeModify, ProxySettings, MarzbanTokenCache
 
 
 async def main():
@@ -215,5 +215,25 @@ async def main():
     # Closing the API client
     await api.close()
 
+
+async def main_with_token_cache():
+    # An example of usage with auto-renewal of a token after its expiration
+    api = MarzbanAPI(base_url="http://marzban-api.com")
+
+    marz_token = MarzbanTokenCache(
+        client=api,
+        username='login', password='password',
+        token_expire_minutes=1440  # DEFAULT VALUE (Optional argument)
+    )
+    # Get a list of users
+    users = await api.get_users(token=await marz_token.get_token(), offset=0, limit=10)
+    print("Users:", users)
+
+    # Get list of nodes
+    nodes = await api.get_nodes(token=await marz_token.get_token())
+    print("Nodes:", nodes)
+
+
+
 # Run the main async function
 asyncio.run(main())

marzban/__init__.py

@@ -1,4 +1,5 @@
 from .api import MarzbanAPI
+from .utils import MarzbanTokenCache
 from .models import (
     Admin, 
     AdminCreate, 
@@ -34,6 +35,7 @@
 __all__ = (
     "__version__",
     "MarzbanAPI",
+    "MarzbanTokenCache",
     "Admin", 
     "AdminCreate", 
     "AdminModify", 

marzban/utils.py

@@ -0,0 +1,35 @@
+import logging
+from datetime import datetime, timedelta
+from typing import Optional
+
+from marzban import MarzbanAPI
+
+
+class MarzbanTokenCache:
+    def __init__(self, client: MarzbanAPI,
+                 username: str, password: str,
+                 token_expire_minutes: int = 1440):
+        self._client = client
+        self._username = username
+        self._password = password
+        self._token_expire_minutes = token_expire_minutes
+        self._token: Optional[str] = None
+        self._exp_at: Optional[datetime] = None
+
+    async def get_token(self):
+        if not self._exp_at or self._exp_at < datetime.now():
+            logging.info(f'Get new token')
+            self._token = await self.get_new_token()
+            self._exp_at = datetime.now() + timedelta(minutes=self._token_expire_minutes - 1)
+        return self._token
+
+    async def get_new_token(self):
+        try:
+            token = await self._client.get_token(
+                username=self._username,
+                password=self._password
+            )
+            return token.access_token
+        except Exception as e:
+            logging.error(f'{e}', exc_info=True)
+            raise e