Source code for q2_smart.ts_endpoints.audience

from q2_smart.exceptions import (
    Q2SmartBadRequest,
    Q2SmartConflict,
    Q2SmartInternalError,
    Q2SmartResourceNotFound,
)

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, IMPORTED, LIST, COUNT


_UNSET = object()




class Audience(TSEndpoint):


    @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()



    @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()



    @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()



    @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()



    @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()



    @authenticate
    async def create_imported_audience(
        self, name: str, service: str, description: str = None
    ) -> dict:
        body = {"name": name, "service": service}
        if description is not None:
            body["description"] = description
        response = await self._send_request(
            f"{TS}/{IMPORTED}/{AUD}",
            CallType.POST,
            mock_response=self.mock_create_imported_audience,
            json=body,
        )
        if response.status_code == 400:
            raise Q2SmartBadRequest(response.json())
        elif response.status_code == 500:
            raise Q2SmartInternalError(response.json())
        return response.json()


    @property
    def mock_create_imported_audience(self) -> dict:
        return audience_data.create_imported_audience_mock()



    @authenticate
    async def update_imported_audience(
        self,
        audience_id: str,
        name: str = None,
        service: str = None,
        description: str = _UNSET,
    ) -> dict:
        body = {}
        if name is not None:
            body["name"] = name
        if service is not None:
            body["service"] = service
        if description is not _UNSET:
            body["description"] = description
        response = await self._send_request(
            f"{TS}/{IMPORTED}/{AUD}/{audience_id}",
            CallType.PATCH,
            mock_response=self.mock_update_imported_audience,
            json=body,
        )
        if response.status_code == 400:
            raise Q2SmartBadRequest(response.json())
        elif response.status_code == 404:
            raise Q2SmartResourceNotFound(response.json())
        elif response.status_code == 500:
            raise Q2SmartInternalError(response.json())
        return response.json()


    @property
    def mock_update_imported_audience(self) -> dict:
        return audience_data.update_imported_audience_mock()



    @authenticate
    async def delete_imported_audience(self, audience_id: str) -> bool:
        response = await self._send_request(
            f"{TS}/{IMPORTED}/{AUD}/{audience_id}",
            CallType.DELETE,
            mock_response=self.mock_delete_imported_audience,
        )
        if response.status_code == 400:
            raise Q2SmartBadRequest(response.json())
        elif response.status_code == 409:
            raise Q2SmartConflict(response.json())
        elif response.status_code == 500:
            raise Q2SmartInternalError(response.json())
        return response.json()


    @property
    def mock_delete_imported_audience(self) -> bool:
        return audience_data.delete_imported_audience_mock()