# This file was auto-generated by Fern from our API Definition.

import typing

from json.decoder import JSONDecodeError

from ...core.api_error import ApiError

from ...core.client_wrapper import AsyncClientWrapper, SyncClientWrapper

from ...core.http_response import AsyncHttpResponse, HttpResponse

from ...core.jsonable_encoder import jsonable_encoder

from ...core.request_options import RequestOptions

from ...core.serialization import convert_and_respect_annotation_metadata

from ...core.unchecked_base_model import construct_type

from ..errors.bad_request_error import BadRequestError

from ..errors.not_found_error import NotFoundError

from ..errors.unauthorized_error import UnauthorizedError

from ..types.error import Error

from ..types.tag_list import TagList

from .types.create_tag_request_body import CreateTagRequestBody

from .types.tag import Tag

# this is used as the default value for optional parameters

OMIT = typing.cast(typing.Any, ...)

class RawTagsClient:

def __init__(self, *, client_wrapper: SyncClientWrapper):

self._client_wrapper = client_wrapper

def attach_tag_to_contact(

self, contact_id: str, *, id: str, request_options: typing.Optional[RequestOptions] = None

) -> HttpResponse[Tag]:

"""

You can tag a specific contact. This will return a tag object for the tag that was added to the contact.

Parameters

----------

contact_id : str

The unique identifier for the contact which is given by Intercom

id : str

The unique identifier for the tag which is given by Intercom

request_options : typing.Optional[RequestOptions]

Request-specific configuration.

Returns

-------

HttpResponse[Tag]

successful

"""

_response = self._client_wrapper.httpx_client.request(

f"contacts/{jsonable_encoder(contact_id)}/tags",

method="POST",

json={

"id": id,

},

headers={

"content-type": "application/json",

},

request_options=request_options,

omit=OMIT,

)

try:

if 200 <= _response.status_code < 300:

_data = typing.cast(

Tag,

construct_type(

type_=Tag, # type: ignore

object_=_response.json(),

),

)

return HttpResponse(response=_response, data=_data)

if _response.status_code == 401:

raise UnauthorizedError(

headers=dict(_response.headers),

body=typing.cast(

Error,

construct_type(

type_=Error, # type: ignore

object_=_response.json(),

),

),

)

if _response.status_code == 404:

raise NotFoundError(

headers=dict(_response.headers),

body=typing.cast(

typing.Any,

construct_type(

type_=typing.Any, # type: ignore

object_=_response.json(),

),

),

)

_response_json = _response.json()

except JSONDecodeError:

raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)

raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)

def detach_tag_from_contact(

self, contact_id: str, id: str, *, request_options: typing.Optional[RequestOptions] = None

) -> HttpResponse[Tag]:

"""

You can remove tag from a specific contact. This will return a tag object for the tag that was removed from the contact.

Parameters

----------

contact_id : str

The unique identifier for the contact which is given by Intercom

id : str

The unique identifier for the tag which is given by Intercom

request_options : typing.Optional[RequestOptions]

Request-specific configuration.

Returns

-------

HttpResponse[Tag]

successful

"""

_response = self._client_wrapper.httpx_client.request(

f"contacts/{jsonable_encoder(contact_id)}/tags/{jsonable_encoder(id)}",

method="DELETE",

request_options=request_options,

)

try:

if 200 <= _response.status_code < 300:

_data = typing.cast(

Tag,

construct_type(

type_=Tag, # type: ignore

object_=_response.json(),

),

)

return HttpResponse(response=_response, data=_data)

if _response.status_code == 401:

raise UnauthorizedError(

headers=dict(_response.headers),

body=typing.cast(

Error,

construct_type(

type_=Error, # type: ignore

object_=_response.json(),

),

),

)

if _response.status_code == 404:

raise NotFoundError(

headers=dict(_response.headers),

body=typing.cast(

typing.Any,

construct_type(

type_=typing.Any, # type: ignore

object_=_response.json(),

),

),

)

_response_json = _response.json()

except JSONDecodeError:

raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)

raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)

def attach_tag_to_conversation(

self, conversation_id: str, *, id: str, admin_id: str, request_options: typing.Optional[RequestOptions] = None

) -> HttpResponse[Tag]:

"""

You can tag a specific conversation. This will return a tag object for the tag that was added to the conversation.

Parameters

----------

conversation_id : str

conversation_id

id : str

The unique identifier for the tag which is given by Intercom

admin_id : str

The unique identifier for the admin which is given by Intercom.

request_options : typing.Optional[RequestOptions]

Request-specific configuration.

Returns

-------

HttpResponse[Tag]

successful

"""

_response = self._client_wrapper.httpx_client.request(

f"conversations/{jsonable_encoder(conversation_id)}/tags",

method="POST",

json={

"id": id,

"admin_id": admin_id,

},

headers={

"content-type": "application/json",

},

request_options=request_options,

omit=OMIT,

)

try:

if 200 <= _response.status_code < 300:

_data = typing.cast(

Tag,

construct_type(

type_=Tag, # type: ignore

object_=_response.json(),

),

)

return HttpResponse(response=_response, data=_data)

if _response.status_code == 401:

raise UnauthorizedError(

headers=dict(_response.headers),

body=typing.cast(

Error,

construct_type(

type_=Error, # type: ignore

object_=_response.json(),

),

),

)

if _response.status_code == 404:

raise NotFoundError(

headers=dict(_response.headers),

body=typing.cast(

typing.Any,

construct_type(

type_=typing.Any, # type: ignore

object_=_response.json(),

),

),

)

_response_json = _response.json()

except JSONDecodeError:

raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)

raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)

def detach_tag_from_conversation(

self, conversation_id: str, id: str, *, admin_id: str, request_options: typing.Optional[RequestOptions] = None

) -> HttpResponse[Tag]:

"""

You can remove tag from a specific conversation. This will return a tag object for the tag that was removed from the conversation.

Parameters

----------

conversation_id : str

conversation_id

id : str

id

admin_id : str

The unique identifier for the admin which is given by Intercom.

request_options : typing.Optional[RequestOptions]

Request-specific configuration.

Returns

-------

HttpResponse[Tag]

successful

"""

_response = self._client_wrapper.httpx_client.request(

f"conversations/{jsonable_encoder(conversation_id)}/tags/{jsonable_encoder(id)}",

method="DELETE",

json={

"admin_id": admin_id,

},

headers={

"content-type": "application/json",

},

request_options=request_options,

omit=OMIT,

)

try:

if 200 <= _response.status_code < 300:

_data = typing.cast(

Tag,

construct_type(

type_=Tag, # type: ignore

object_=_response.json(),

),

)

return HttpResponse(response=_response, data=_data)

if _response.status_code == 401:

raise UnauthorizedError(

headers=dict(_response.headers),

body=typing.cast(

Error,

construct_type(

type_=Error, # type: ignore

object_=_response.json(),

),

),

)

if _response.status_code == 404:

raise NotFoundError(

headers=dict(_response.headers),

body=typing.cast(

typing.Any,

construct_type(

type_=typing.Any, # type: ignore

object_=_response.json(),

),

),

)

_response_json = _response.json()

except JSONDecodeError:

raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)

raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)

def list_tags(self, *, request_options: typing.Optional[RequestOptions] = None) -> HttpResponse[TagList]:

"""

You can fetch a list of all tags for a given workspace.

Parameters

----------

request_options : typing.Optional[RequestOptions]

Request-specific configuration.

Returns

-------

HttpResponse[TagList]

successful

"""

_response = self._client_wrapper.httpx_client.request(

"tags",

method="GET",

request_options=request_options,

)

try:

if 200 <= _response.status_code < 300:

_data = typing.cast(

TagList,

construct_type(

type_=TagList, # type: ignore

object_=_response.json(),

),

)

return HttpResponse(response=_response, data=_data)

if _response.status_code == 401:

raise UnauthorizedError(

headers=dict(_response.headers),

body=typing.cast(

Error,

construct_type(

type_=Error, # type: ignore

object_=_response.json(),

),

),

)

_response_json = _response.json()

except JSONDecodeError:

raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)

raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)

def create_tag(

self, *, request: CreateTagRequestBody, request_options: typing.Optional[RequestOptions] = None

) -> HttpResponse[Tag]:

"""

You can use this endpoint to perform the following operations:

**1. Create a new tag:** You can create a new tag by passing in the tag name as specified in "Create or Update Tag Request Payload" described below.

**2. Update an existing tag:** You can update an existing tag by passing the id of the tag as specified in "Create or Update Tag Request Payload" described below.

**3. Tag Companies:** You can tag single company or a list of companies. You can tag a company by passing in the tag name and the company details as specified in "Tag Company Request Payload" described below. Also, if the tag doesn't exist then a new one will be created automatically.

**4. Untag Companies:** You can untag a single company or a list of companies. You can untag a company by passing in the tag id and the company details as specified in "Untag Company Request Payload" described below.

**5. Tag Multiple Users:** You can tag a list of users. You can tag the users by passing in the tag name and the user details as specified in "Tag Users Request Payload" described below.

Each operation will return a tag object.

Parameters

----------

request : CreateTagRequestBody

request_options : typing.Optional[RequestOptions]

Request-specific configuration.

Returns

-------

HttpResponse[Tag]

Action successful

"""

_response = self._client_wrapper.httpx_client.request(

"tags",

method="POST",

json=convert_and_respect_annotation_metadata(

object_=request, annotation=CreateTagRequestBody, direction="write"

),

headers={

"content-type": "application/json",

},

request_options=request_options,

omit=OMIT,

)

try:

if 200 <= _response.status_code < 300:

_data = typing.cast(

Tag,

construct_type(

type_=Tag, # type: ignore

object_=_response.json(),

),

)

return HttpResponse(response=_response, data=_data)

if _response.status_code == 400:

raise BadRequestError(

headers=dict(_response.headers),

body=typing.cast(

typing.Any,

construct_type(

type_=typing.Any, # type: ignore

object_=_response.json(),

),

),

)

if _response.status_code == 401:

raise UnauthorizedError(

headers=dict(_response.headers),

body=typing.cast(

Error,

construct_type(

type_=Error, # type: ignore

object_=_response.json(),

),

),

)

if _response.status_code == 404:

raise NotFoundError(

headers=dict(_response.headers),

body=typing.cast(

typing.Any,

construct_type(

type_=typing.Any, # type: ignore

object_=_response.json(),

),

),

)

_response_json = _response.json()

except JSONDecodeError:

raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)

raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)

def find_tag(self, id: str, *, request_options: typing.Optional[RequestOptions] = None) -> HttpResponse[Tag]:

"""

You can fetch the details of tags that are on the workspace by their id.

This will return a tag object.

Parameters

----------

id : str

The unique identifier of a given tag

request_options : typing.Optional[RequestOptions]

Request-specific configuration.

Returns

-------

HttpResponse[Tag]

Tag found

"""

_response = self._client_wrapper.httpx_client.request(

f"tags/{jsonable_encoder(id)}",

method="GET",

request_options=request_options,

)

try:

if 200 <= _response.status_code < 300:

_data = typing.cast(

Tag,

construct_type(

type_=Tag, # type: ignore

object_=_response.json(),

),

)

return HttpResponse(response=_response, data=_data)

if _response.status_code == 401:

raise UnauthorizedError(

headers=dict(_response.headers),

body=typing.cast(

Error,

construct_type(

type_=Error, # type: ignore

object_=_response.json(),

),

),

)

if _response.status_code == 404:

raise NotFoundError(

headers=dict(_response.headers),

body=typing.cast(

typing.Any,

construct_type(

type_=typing.Any, # type: ignore

object_=_response.json(),

),

),

)

_response_json = _response.json()

except JSONDecodeError:

raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)

raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)

def delete_tag(self, id: str, *, request_options: typing.Optional[RequestOptions] = None) -> HttpResponse[None]:

"""

You can delete the details of tags that are on the workspace by passing in the id.

Parameters

----------

id : str

The unique identifier of a given tag

request_options : typing.Optional[RequestOptions]

Request-specific configuration.

Returns

-------

HttpResponse[None]

"""

_response = self._client_wrapper.httpx_client.request(

f"tags/{jsonable_encoder(id)}",

method="DELETE",

request_options=request_options,

)

try:

if 200 <= _response.status_code < 300:

return HttpResponse(response=_response, data=None)

if _response.status_code == 400:

raise BadRequestError(

headers=dict(_response.headers),

body=typing.cast(

typing.Any,

construct_type(

type_=typing.Any, # type: ignore

object_=_response.json(),

),

),

)

if _response.status_code == 401:

raise UnauthorizedError(

headers=dict(_response.headers),

body=typing.cast(

Error,

construct_type(

type_=Error, # type: ignore

object_=_response.json(),

),

),

)

if _response.status_code == 404:

raise NotFoundError(

headers=dict(_response.headers),

body=typing.cast(

typing.Any,

construct_type(

type_=typing.Any, # type: ignore

object_=_response.json(),

),

),

)

_response_json = _response.json()

except JSONDecodeError:

raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)

raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)

def attach_tag_to_ticket(

self, ticket_id: str, *, id: str, admin_id: str, request_options: typing.Optional[RequestOptions] = None

) -> HttpResponse[Tag]:

"""

You can tag a specific ticket. This will return a tag object for the tag that was added to the ticket.

Parameters

----------

ticket_id : str

ticket_id

id : str

The unique identifier for the tag which is given by Intercom

admin_id : str

The unique identifier for the admin which is given by Intercom.

request_options : typing.Optional[RequestOptions]

Request-specific configuration.

Returns

-------

HttpResponse[Tag]

successful

"""

_response = self._client_wrapper.httpx_client.request(

f"tickets/{jsonable_encoder(ticket_id)}/tags",

method="POST",

json={

"id": id,

"admin_id": admin_id,

},

headers={

"content-type": "application/json",

},

request_options=request_options,

omit=OMIT,

)

try:

if 200 <= _response.status_code < 300:

_data = typing.cast(

Tag,

construct_type(

type_=Tag, # type: ignore

object_=_response.json(),

),

)

return HttpResponse(response=_response, data=_data)

if _response.status_code == 401:

raise UnauthorizedError(

headers=dict(_response.headers),

body=typing.cast(

Error,

construct_type(

type_=Error, # type: ignore

object_=_response.json(),

),

),

)

if _response.status_code == 404:

raise NotFoundError(

headers=dict(_response.headers),

body=typing.cast(

typing.Any,

construct_type(

type_=typing.Any, # type: ignore

object_=_response.json(),

),

),

)

_response_json = _response.json()

except JSONDecodeError:

raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)

raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)

def detach_tag_from_ticket(

self, ticket_id: str, id: str, *, admin_id: str, request_options: typing.Optional[RequestOptions] = None

) -> HttpResponse[Tag]:

"""

You can remove tag from a specific ticket. This will return a tag object for the tag that was removed from the ticket.

Parameters

----------

ticket_id : str

ticket_id

id : str

The unique identifier for the tag which is given by Intercom

admin_id : str

The unique identifier for the admin which is given by Intercom.

request_options : typing.Optional[RequestOptions]

Request-specific configuration.

Returns

-------

HttpResponse[Tag]

successful

"""

_response = self._client_wrapper.httpx_client.request(

f"tickets/{jsonable_encoder(ticket_id)}/tags/{jsonable_encoder(id)}",

method="DELETE",

json={

"admin_id": admin_id,

},

headers={

"content-type": "application/json",

},

request_options=request_options,

omit=OMIT,

)

try:

if 200 <= _response.status_code < 300:

_data = typing.cast(

Tag,

construct_type(

type_=Tag, # type: ignore

object_=_response.json(),

),

)

return HttpResponse(response=_response, data=_data)

if _response.status_code == 401:

raise UnauthorizedError(

headers=dict(_response.headers),

body=typing.cast(

Error,

construct_type(

type_=Error, # type: ignore

object_=_response.json(),

),

),

)

if _response.status_code == 404:

raise NotFoundError(

headers=dict(_response.headers),

body=typing.cast(

typing.Any,

construct_type(

type_=typing.Any, # type: ignore

object_=_response.json(),

),

),

)

_response_json = _response.json()

except JSONDecodeError:

raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)

raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)

class AsyncRawTagsClient:

def __init__(self, *, client_wrapper: AsyncClientWrapper):

self._client_wrapper = client_wrapper

async def attach_tag_to_contact(

self, contact_id: str, *, id: str, request_options: typing.Optional[RequestOptions] = None

) -> AsyncHttpResponse[Tag]:

"""

You can tag a specific contact. This will return a tag object for the tag that was added to the contact.

Parameters

----------

contact_id : str

The unique identifier for the contact which is given by Intercom

id : str

The unique identifier for the tag which is given by Intercom

request_options : typing.Optional[RequestOptions]

Request-specific configuration.

Returns

-------

AsyncHttpResponse[Tag]

successful

"""

_response = await self._client_wrapper.httpx_client.request(

f"contacts/{jsonable_encoder(contact_id)}/tags",

method="POST",

json={

"id": id,

},

headers={

"content-type": "application/json",

},

request_options=request_options,

omit=OMIT,

)

try:

if 200 <= _response.status_code < 300:

_data = typing.cast(

Tag,

construct_type(

type_=Tag, # type: ignore

object_=_response.json(),

),

)

return AsyncHttpResponse(response=_response, data=_data)

if _response.status_code == 401:

raise UnauthorizedError(

headers=dict(_response.headers),

body=typing.cast(

Error,

construct_type(

type_=Error, # type: ignore

object_=_response.json(),

),

),

)

if _response.status_code == 404:

raise NotFoundError(

headers=dict(_response.headers),

body=typing.cast(

typing.Any,

construct_type(

type_=typing.Any, # type: ignore

object_=_response.json(),

),

),

)

_response_json = _response.json()

except JSONDecodeError:

raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)

raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)

async def detach_tag_from_contact(

self, contact_id: str, id: str, *, request_options: typing.Optional[RequestOptions] = None

) -> AsyncHttpResponse[Tag]:

"""

You can remove tag from a specific contact. This will return a tag object for the tag that was removed from the contact.

Parameters

----------

contact_id : str

The unique identifier for the contact which is given by Intercom

id : str

The unique identifier for the tag which is given by Intercom

request_options : typing.Optional[RequestOptions]

Request-specific configuration.

Returns

-------

AsyncHttpResponse[Tag]

successful

"""

_response = await self._client_wrapper.httpx_client.request(

f"contacts/{jsonable_encoder(contact_id)}/tags/{jsonable_encoder(id)}",

method="DELETE",

request_options=request_options,

)

try:

if 200 <= _response.status_code < 300:

_data = typing.cast(

Tag,

construct_type(

type_=Tag, # type: ignore

object_=_response.json(),

),

)

return AsyncHttpResponse(response=_response, data=_data)

if _response.status_code == 401:

raise UnauthorizedError(

headers=dict(_response.headers),

body=typing.cast(

Error,

construct_type(

type_=Error, # type: ignore

object_=_response.json(),

),

),

)

if _response.status_code == 404:

raise NotFoundError(

headers=dict(_response.headers),

body=typing.cast(

typing.Any,

construct_type(

type_=typing.Any, # type: ignore

object_=_response.json(),

),

),

)

_response_json = _response.json()

except JSONDecodeError:

raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)

raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)

async def attach_tag_to_conversation(

self, conversation_id: str, *, id: str, admin_id: str, request_options: typing.Optional[RequestOptions] = None

) -> AsyncHttpResponse[Tag]:

"""

You can tag a specific conversation. This will return a tag object for the tag that was added to the conversation.

Parameters

----------

conversation_id : str

conversation_id

id : str

The unique identifier for the tag which is given by Intercom

admin_id : str

The unique identifier for the admin which is given by Intercom.

request_options : typing.Optional[RequestOptions]

Request-specific configuration.

Returns

-------

AsyncHttpResponse[Tag]

successful

"""

_response = await self._client_wrapper.httpx_client.request(

f"conversations/{jsonable_encoder(conversation_id)}/tags",

method="POST",

json={

"id": id,

"admin_id": admin_id,

},

headers={

"content-type": "application/json",

},

request_options=request_options,

omit=OMIT,

)

try:

if 200 <= _response.status_code < 300:

_data = typing.cast(

Tag,

construct_type(

type_=Tag, # type: ignore

object_=_response.json(),

),

)

return AsyncHttpResponse(response=_response, data=_data)

if _response.status_code == 401:

raise UnauthorizedError(

headers=dict(_response.headers),

body=typing.cast(

Error,

construct_type(

type_=Error, # type: ignore

object_=_response.json(),

),

),

)

if _response.status_code == 404:

raise NotFoundError(

headers=dict(_response.headers),

body=typing.cast(

typing.Any,

construct_type(

type_=typing.Any, # type: ignore

object_=_response.json(),

),

),

)

_response_json = _response.json()

except JSONDecodeError:

raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)

raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)

async def detach_tag_from_conversation(

self, conversation_id: str, id: str, *, admin_id: str, request_options: typing.Optional[RequestOptions] = None

) -> AsyncHttpResponse[Tag]:

"""

You can remove tag from a specific conversation. This will return a tag object for the tag that was removed from the conversation.

Parameters

----------

conversation_id : str

conversation_id

id : str

id

admin_id : str

The unique identifier for the admin which is given by Intercom.

request_options : typing.Optional[RequestOptions]

Request-specific configuration.

Returns

-------

AsyncHttpResponse[Tag]

successful

"""

_response = await self._client_wrapper.httpx_client.request(

f"conversations/{jsonable_encoder(conversation_id)}/tags/{jsonable_encoder(id)}",

method="DELETE",

json={

"admin_id": admin_id,

},

headers={

"content-type": "application/json",

},

request_options=request_options,

omit=OMIT,

)

try:

if 200 <= _response.status_code < 300:

_data = typing.cast(

Tag,

construct_type(

type_=Tag, # type: ignore

object_=_response.json(),

),

)

return AsyncHttpResponse(response=_response, data=_data)

if _response.status_code == 401:

raise UnauthorizedError(

headers=dict(_response.headers),

body=typing.cast(

Error,

construct_type(

type_=Error, # type: ignore

object_=_response.json(),

),

),

)

if _response.status_code == 404:

raise NotFoundError(

headers=dict(_response.headers),

body=typing.cast(

typing.Any,

construct_type(

type_=typing.Any, # type: ignore

object_=_response.json(),

),