62 lines
No EOL
2.4 KiB
Python
62 lines
No EOL
2.4 KiB
Python
import uuid
|
|
from enum import Enum
|
|
from datetime import datetime
|
|
|
|
from typing import List, Dict, Any, Optional
|
|
|
|
import aiohttp
|
|
|
|
from .data_models import *
|
|
|
|
|
|
class AsteriskManagerClient:
|
|
def __init__(self, base_url: str, auth_key: Optional[str] = None, token: Optional[str] = None):
|
|
self.base_url = base_url.rstrip('/')
|
|
self.token: Optional[str] = None
|
|
if token:
|
|
self.token = token
|
|
|
|
if auth_key and not token:
|
|
self.auth_key = auth_key
|
|
else:
|
|
self.auth_key = None
|
|
|
|
if not self.token and not self.auth_key:
|
|
raise ValueError("Either token or auth_key must be provided")
|
|
|
|
|
|
async def _request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]:
|
|
headers = kwargs.get('headers', {})
|
|
if self.token:
|
|
headers['Authorization'] = f'Bearer {self.token}'
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.request(method, f"{self.base_url}{endpoint}", **kwargs) as response:
|
|
if response.status >= 400:
|
|
error_data = await response.json()
|
|
raise Exception(f"API Error {response.status}: {error_data}")
|
|
return await response.json()
|
|
|
|
|
|
async def register_manager(self, auth_key: Optional[str] = None) -> AccessResponse:
|
|
if auth_key is None and self.auth_key is None:
|
|
raise ValueError("Auth key is required for registration")
|
|
auth_key = auth_key or self.auth_key
|
|
data = Registration(auth_key=auth_key).dict()
|
|
response = await self._request('POST', '/asterisk_manager/register', json=data)
|
|
access_response = AccessResponse(**response)
|
|
self.token = access_response.token
|
|
return access_response
|
|
|
|
|
|
async def get_sip_accounts(self, **params) -> List[SIPAccount]:
|
|
response = await self._request('GET', '/sip/', params=params)
|
|
return [SIPAccount(**account) for account in response]
|
|
|
|
async def get_sip_account(self, account_id: int) -> SIPAccount:
|
|
response = await self._request('GET', f'/sip/{account_id}')
|
|
return SIPAccount(**response)
|
|
|
|
async def set_sip_register_status(self, account_id: int, is_registered: bool):
|
|
data = {"is_registered": is_registered}
|
|
response = await self._request('PATCH', f'/sip/{account_id}', json=data)
|
|
return SIPAccount(**response) |