|
@@ -1,11 +1,20 @@
|
|
import requests
|
|
import requests
|
|
import json
|
|
import json
|
|
from datetime import datetime
|
|
from datetime import datetime
|
|
|
|
+from typing import List, Dict, Any
|
|
|
|
|
|
|
|
|
|
-def get_urls(base_url, auth: str) -> list:
|
|
|
|
- url = f"{base_url}/getUserAccounts/streams"
|
|
|
|
|
|
+def get_urls(base_url: str, auth: str) -> List[Dict[str, Any]]:
|
|
|
|
+ """Retrieve user account streams from the specified base URL.
|
|
|
|
+
|
|
|
|
+ Args:
|
|
|
|
+ base_url (str): The base URL of the API.
|
|
|
|
+ auth (str): The authorization token for accessing the API.
|
|
|
|
|
|
|
|
+ Returns:
|
|
|
|
+ List[Dict[str, Any]]: A list of user account streams.
|
|
|
|
+ """
|
|
|
|
+ url = f"{base_url}/getUserAccounts/streams"
|
|
payload = {}
|
|
payload = {}
|
|
headers = {"Authorization": f"Basic {auth}"}
|
|
headers = {"Authorization": f"Basic {auth}"}
|
|
|
|
|
|
@@ -13,67 +22,86 @@ def get_urls(base_url, auth: str) -> list:
|
|
return json.loads(response.text)
|
|
return json.loads(response.text)
|
|
|
|
|
|
|
|
|
|
-def get_user_accounts(base_url, auth: str) -> list:
|
|
|
|
- url = f"{base_url}/getUserAccounts"
|
|
|
|
|
|
+def get_user_accounts(base_url: str, auth: str) -> List[Dict[str, Any]]:
|
|
|
|
+ """Retrieve user accounts from the specified base URL.
|
|
|
|
+
|
|
|
|
+ Args:
|
|
|
|
+ base_url (str): The base URL of the API.
|
|
|
|
+ auth (str): The authorization token for accessing the API.
|
|
|
|
|
|
|
|
+ Returns:
|
|
|
|
+ List[Dict[str, Any]]: A list of user accounts with their expiration dates rendered.
|
|
|
|
+ """
|
|
|
|
+ url = f"{base_url}/getUserAccounts"
|
|
payload = {}
|
|
payload = {}
|
|
headers = {"Authorization": f"Basic {auth}"}
|
|
headers = {"Authorization": f"Basic {auth}"}
|
|
|
|
|
|
response = requests.request("GET", url, headers=headers, data=payload)
|
|
response = requests.request("GET", url, headers=headers, data=payload)
|
|
res_json = json.loads(response.text)
|
|
res_json = json.loads(response.text)
|
|
|
|
+
|
|
for account in res_json:
|
|
for account in res_json:
|
|
account["expiaryDate_rendered"] = datetime.utcfromtimestamp(
|
|
account["expiaryDate_rendered"] = datetime.utcfromtimestamp(
|
|
account["expiaryDate"]
|
|
account["expiaryDate"]
|
|
).strftime("%d/%m/%Y")
|
|
).strftime("%d/%m/%Y")
|
|
|
|
+
|
|
return res_json
|
|
return res_json
|
|
|
|
|
|
|
|
|
|
-def delete_user_account(base_url: str, auth: str, stream:str, username:str) -> bool:
|
|
|
|
- """_summary_
|
|
|
|
|
|
+def delete_user_account(base_url: str, auth: str, stream: str, username: str) -> bool:
|
|
|
|
+ """Delete a user account from the specified base URL.
|
|
|
|
|
|
Args:
|
|
Args:
|
|
- base_url (str): _description_
|
|
|
|
- auth (str): _description_
|
|
|
|
- stream (str): _description_
|
|
|
|
- username (str): _description_
|
|
|
|
|
|
+ base_url (str): The base URL of the API.
|
|
|
|
+ auth (str): The authorization token for accessing the API.
|
|
|
|
+ stream (str): The name of the stream associated with the user account.
|
|
|
|
+ username (str): The username of the account to delete.
|
|
|
|
|
|
Returns:
|
|
Returns:
|
|
- bool: _description_
|
|
|
|
- """
|
|
|
|
|
|
+ bool: True if the account was deleted successfully, False otherwise.
|
|
|
|
+ """
|
|
url = f"{base_url}/deleteAccount"
|
|
url = f"{base_url}/deleteAccount"
|
|
-
|
|
|
|
payload = {"stream": stream, "user": username}
|
|
payload = {"stream": stream, "user": username}
|
|
-
|
|
|
|
headers = {"Authorization": f"Basic {auth}"}
|
|
headers = {"Authorization": f"Basic {auth}"}
|
|
|
|
|
|
response = requests.request("POST", url, headers=headers, data=payload)
|
|
response = requests.request("POST", url, headers=headers, data=payload)
|
|
- if "Deleted" in response.text:
|
|
|
|
- return True
|
|
|
|
- else:
|
|
|
|
- return False
|
|
|
|
|
|
+ return "Deleted" in response.text
|
|
|
|
|
|
|
|
|
|
-
|
|
|
|
-def add_user_account(base_url: str, auth: str, username:str, password:str, stream:str,) -> bool:
|
|
|
|
- """_summary_
|
|
|
|
|
|
+def add_user_account(base_url: str, auth: str, username: str, password: str, stream: str) -> bool:
|
|
|
|
+ """Add a user account to the specified base URL.
|
|
|
|
|
|
Args:
|
|
Args:
|
|
- base_url (str): _description_
|
|
|
|
- auth (str): _description_
|
|
|
|
- stream (str): _description_
|
|
|
|
- username (str): _description_
|
|
|
|
|
|
+ base_url (str): The base URL of the API.
|
|
|
|
+ auth (str): The authorization token for accessing the API.
|
|
|
|
+ username (str): The username of the account to add.
|
|
|
|
+ password (str): The password of the account to add.
|
|
|
|
+ stream (str): The name of the stream associated with the user account.
|
|
|
|
|
|
Returns:
|
|
Returns:
|
|
- bool: _description_
|
|
|
|
- """
|
|
|
|
|
|
+ bool: True if the account was added successfully, False otherwise.
|
|
|
|
+ """
|
|
url = f"{base_url}/addAccount"
|
|
url = f"{base_url}/addAccount"
|
|
-
|
|
|
|
payload = {"username": username, "password": password, "stream": stream}
|
|
payload = {"username": username, "password": password, "stream": stream}
|
|
-
|
|
|
|
headers = {"Authorization": f"Basic {auth}"}
|
|
headers = {"Authorization": f"Basic {auth}"}
|
|
|
|
|
|
response = requests.request("POST", url, headers=headers, data=payload)
|
|
response = requests.request("POST", url, headers=headers, data=payload)
|
|
- if "Added successfully" in response.text:
|
|
|
|
- return True
|
|
|
|
- else:
|
|
|
|
- return False
|
|
|
|
|
|
+ return "Added successfully" in response.text
|
|
|
|
+
|
|
|
|
+
|
|
|
|
+def get_user_accounts_count(base_url: str, auth: str) -> int:
|
|
|
|
+ """Get the count of user accounts from the specified base URL.
|
|
|
|
+
|
|
|
|
+ Args:
|
|
|
|
+ base_url (str): The base URL of the API.
|
|
|
|
+ auth (str): The authorization token for accessing the API.
|
|
|
|
+
|
|
|
|
+ Returns:
|
|
|
|
+ int: The count of user accounts.
|
|
|
|
+ """
|
|
|
|
+ url = f"{base_url}/getUserAccounts/count"
|
|
|
|
+ payload = {}
|
|
|
|
+ headers = {"Authorization": f"Basic {auth}"}
|
|
|
|
+
|
|
|
|
+ response = requests.request("GET", url, headers=headers, data=payload)
|
|
|
|
+ res_json = json.loads(response.text)
|
|
|
|
+ return res_json['count']
|