85 lines
2.8 KiB
Python
85 lines
2.8 KiB
Python
import requests
|
|
import json
|
|
from datetime import datetime
|
|
from typing import List, Dict, Any, Optional
|
|
|
|
# Create a session object to reuse TCP connections
|
|
session = requests.Session()
|
|
|
|
def _make_api_request(
|
|
method: str,
|
|
base_url: str,
|
|
auth: str,
|
|
endpoint: str,
|
|
payload: Optional[Dict[str, Any]] = None,
|
|
) -> Any:
|
|
"""
|
|
A helper function to make API requests.
|
|
|
|
Args:
|
|
method: The HTTP method to use (e.g., 'GET', 'POST').
|
|
base_url: The base URL of the API.
|
|
auth: The authorization token.
|
|
endpoint: The API endpoint to call.
|
|
payload: The data to send with the request.
|
|
|
|
Returns:
|
|
The JSON response from the API.
|
|
"""
|
|
url = f"{base_url}/{endpoint}"
|
|
headers = {"Authorization": f"Basic {auth}"}
|
|
try:
|
|
response = session.request(method, url, headers=headers, data=payload)
|
|
response.raise_for_status()
|
|
return response
|
|
except (requests.exceptions.RequestException, json.JSONDecodeError) as e:
|
|
# Log the error for debugging purposes
|
|
print(f"API request failed: {e}")
|
|
return None
|
|
|
|
|
|
def get_urls(base_url: str, auth: str) -> List[Dict[str, Any]]:
|
|
"""Retrieves user account streams from the API."""
|
|
response = _make_api_request("GET", base_url, auth, "getUserAccounts/streams")
|
|
return response.json() if response else []
|
|
|
|
|
|
def get_user_accounts(base_url: str, auth: str) -> List[Dict[str, Any]]:
|
|
"""Retrieves user accounts from the API."""
|
|
response = _make_api_request("GET", base_url, auth, "getUserAccounts")
|
|
if not response:
|
|
return []
|
|
|
|
accounts = response.json()
|
|
for account in accounts:
|
|
account["expiaryDate_rendered"] = datetime.utcfromtimestamp(
|
|
account["expiaryDate"]
|
|
).strftime("%d/%m/%Y")
|
|
return accounts
|
|
|
|
|
|
def delete_user_account(base_url: str, auth: str, stream: str, username: str) -> bool:
|
|
"""Deletes a user account via the API."""
|
|
payload = {"stream": stream, "user": username}
|
|
response = _make_api_request(
|
|
"POST", base_url, auth, "deleteAccount", payload=payload
|
|
)
|
|
return response and "Deleted" in response.text
|
|
|
|
|
|
def add_user_account(
|
|
base_url: str, auth: str, username: str, password: str, stream: str
|
|
) -> bool:
|
|
"""Adds a user account via the API."""
|
|
payload = {"username": username, "password": password, "stream": stream}
|
|
response = _make_api_request(
|
|
"POST", base_url, auth, "addAccount", payload=payload
|
|
)
|
|
return response and response.status_code == 200
|
|
|
|
|
|
def get_stream_names(base_url: str, auth: str) -> List[str]:
|
|
"""Retrieves a list of stream names from the API."""
|
|
response = _make_api_request("GET", base_url, auth, "getStreamNames")
|
|
return response.json() if response else []
|