from q2_smart.exceptions import Q2SmartBadRequest, Q2SmartInternalError
from .base import CallType, TSEndpoint
from .models.audience_list import AudienceList
from .decorators import authenticate
from .mocks import audience_data
from .constants import TS, AUD, LIST, COUNT
[docs]
class Audience(TSEndpoint):
[docs]
@authenticate
async def get(self) -> AudienceList:
response = await self._send_request(
f"{TS}/{AUD}", CallType.GET, mock_response=self.mock_get_audiences
)
return AudienceList.from_list(response.json())
@property
def mock_get_audiences(self) -> dict:
return audience_data.get_audience_list_mock()
[docs]
@authenticate
async def get_users(self, audience_id: str) -> list[int]:
response = await self._send_request(
f"{TS}/{AUD}/{audience_id}/{LIST}",
CallType.GET,
mock_response=self.mock_audience_users,
)
self._check_response_status(response, f"Audience {audience_id} does not exist")
return response.json()
@property
def mock_audience_users(self) -> list[int]:
return audience_data.get_audience_users_mock()
[docs]
@authenticate
async def get_count(self, audience_id: str) -> int:
response = await self._send_request(
f"{TS}/{AUD}/{audience_id}/{COUNT}",
CallType.GET,
mock_response=self.mock_audience_count,
)
self._check_response_status(response, f"Audience {audience_id} does not exist")
return int(response.text)
@property
def mock_audience_count(self) -> int:
return audience_data.get_audience_users_count_mock()
[docs]
@authenticate
async def create_simplified_audience(
self, name: str, service: str, query: dict, description: str = None
) -> dict:
body = {"name": name, "service": service, "query": query}
if description is not None:
body["description"] = description
response = await self._send_request(
f"{TS}/{AUD}",
CallType.POST,
mock_response=self.mock_create_audience,
json=body,
)
if response.status_code == 500:
raise Q2SmartInternalError(response.json())
if response.status_code == 400:
raise Q2SmartBadRequest(response.json())
return response.json()
@property
def mock_create_audience(self) -> dict:
return audience_data.create_audience_mock()
[docs]
@authenticate
async def delete_simplified_audience(self, audience_id: str) -> bool:
response = await self._send_request(
f"{TS}/{AUD}/{audience_id}",
CallType.DELETE,
mock_response=self.mock_delete_audience,
)
if response.status_code == 500:
raise Q2SmartInternalError(response.json())
if response.status_code == 400:
raise Q2SmartBadRequest(response.json())
return response.json()
@property
def mock_delete_audience(self) -> bool:
return audience_data.delete_audience_mock()