# pylint: disable=too-many-lines,too-many-statements
# coding=utf-8
# --------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is regenerated.
# --------------------------------------------------------------------------
from io import IOBase
from typing import Any, Callable, Dict, IO, Iterable, Optional, TypeVar, Union, cast, overload
import urllib.parse
from azure.core.exceptions import (
ClientAuthenticationError,
HttpResponseError,
ResourceExistsError,
ResourceNotFoundError,
ResourceNotModifiedError,
map_error,
)
from azure.core.paging import ItemPaged
from azure.core.pipeline import PipelineResponse
from azure.core.pipeline.transport import HttpResponse
from azure.core.polling import LROPoller, NoPolling, PollingMethod
from azure.core.rest import HttpRequest
from azure.core.tracing.decorator import distributed_trace
from azure.core.utils import case_insensitive_dict
from azure.mgmt.core.exceptions import ARMErrorFormat
from azure.mgmt.core.polling.arm_polling import ARMPolling
from .. import models as _models
from .._serialization import Serializer
from .._vendor import _convert_request
T = TypeVar("T")
ClsType = Optional[Callable[[PipelineResponse[HttpRequest, HttpResponse], T, Dict[str, Any]], Any]]
_SERIALIZER = Serializer()
_SERIALIZER.client_side_validation = False
def build_check_name_availability_request(support_ticket_name: str, subscription_id: str, **kwargs: Any) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2024-04-01"))
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = kwargs.pop(
"template_url",
"/subscriptions/{subscriptionId}/providers/Microsoft.Support/supportTickets/{supportTicketName}/checkNameAvailability",
) # pylint: disable=line-too-long
path_format_arguments = {
"supportTicketName": _SERIALIZER.url("support_ticket_name", support_ticket_name, "str"),
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
# Construct headers
if content_type is not None:
_headers["Content-Type"] = _SERIALIZER.header("content_type", content_type, "str")
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="POST", url=_url, params=_params, headers=_headers, **kwargs)
def build_list_request(
support_ticket_name: str,
subscription_id: str,
*,
top: Optional[int] = None,
filter: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2024-04-01"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = kwargs.pop(
"template_url",
"/subscriptions/{subscriptionId}/providers/Microsoft.Support/supportTickets/{supportTicketName}/communications",
) # pylint: disable=line-too-long
path_format_arguments = {
"supportTicketName": _SERIALIZER.url("support_ticket_name", support_ticket_name, "str"),
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
if top is not None:
_params["$top"] = _SERIALIZER.query("top", top, "int")
if filter is not None:
_params["$filter"] = _SERIALIZER.query("filter", filter, "str")
_params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
# Construct headers
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs)
def build_get_request(
support_ticket_name: str, communication_name: str, subscription_id: str, **kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2024-04-01"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = kwargs.pop(
"template_url",
"/subscriptions/{subscriptionId}/providers/Microsoft.Support/supportTickets/{supportTicketName}/communications/{communicationName}",
) # pylint: disable=line-too-long
path_format_arguments = {
"supportTicketName": _SERIALIZER.url("support_ticket_name", support_ticket_name, "str"),
"communicationName": _SERIALIZER.url("communication_name", communication_name, "str"),
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
# Construct headers
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs)
def build_create_request(
support_ticket_name: str, communication_name: str, subscription_id: str, **kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2024-04-01"))
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = kwargs.pop(
"template_url",
"/subscriptions/{subscriptionId}/providers/Microsoft.Support/supportTickets/{supportTicketName}/communications/{communicationName}",
) # pylint: disable=line-too-long
path_format_arguments = {
"supportTicketName": _SERIALIZER.url("support_ticket_name", support_ticket_name, "str"),
"communicationName": _SERIALIZER.url("communication_name", communication_name, "str"),
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
# Construct headers
if content_type is not None:
_headers["Content-Type"] = _SERIALIZER.header("content_type", content_type, "str")
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="PUT", url=_url, params=_params, headers=_headers, **kwargs)
[docs]class CommunicationsOperations:
"""
.. warning::
**DO NOT** instantiate this class directly.
Instead, you should access the following operations through
:class:`~azure.mgmt.support.MicrosoftSupport`'s
:attr:`communications` attribute.
"""
models = _models
def __init__(self, *args, **kwargs):
input_args = list(args)
self._client = input_args.pop(0) if input_args else kwargs.pop("client")
self._config = input_args.pop(0) if input_args else kwargs.pop("config")
self._serialize = input_args.pop(0) if input_args else kwargs.pop("serializer")
self._deserialize = input_args.pop(0) if input_args else kwargs.pop("deserializer")
@overload
def check_name_availability(
self,
support_ticket_name: str,
check_name_availability_input: _models.CheckNameAvailabilityInput,
*,
content_type: str = "application/json",
**kwargs: Any
) -> _models.CheckNameAvailabilityOutput:
"""Check the availability of a resource name. This API should be used to check the uniqueness of
the name for adding a new communication to the support ticket.
:param support_ticket_name: Support ticket name. Required.
:type support_ticket_name: str
:param check_name_availability_input: Input to check. Required.
:type check_name_availability_input: ~azure.mgmt.support.models.CheckNameAvailabilityInput
:keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
Default value is "application/json".
:paramtype content_type: str
:return: CheckNameAvailabilityOutput or the result of cls(response)
:rtype: ~azure.mgmt.support.models.CheckNameAvailabilityOutput
:raises ~azure.core.exceptions.HttpResponseError:
"""
@overload
def check_name_availability(
self,
support_ticket_name: str,
check_name_availability_input: IO[bytes],
*,
content_type: str = "application/json",
**kwargs: Any
) -> _models.CheckNameAvailabilityOutput:
"""Check the availability of a resource name. This API should be used to check the uniqueness of
the name for adding a new communication to the support ticket.
:param support_ticket_name: Support ticket name. Required.
:type support_ticket_name: str
:param check_name_availability_input: Input to check. Required.
:type check_name_availability_input: IO[bytes]
:keyword content_type: Body Parameter content-type. Content type parameter for binary body.
Default value is "application/json".
:paramtype content_type: str
:return: CheckNameAvailabilityOutput or the result of cls(response)
:rtype: ~azure.mgmt.support.models.CheckNameAvailabilityOutput
:raises ~azure.core.exceptions.HttpResponseError:
"""
[docs] @distributed_trace
def check_name_availability(
self,
support_ticket_name: str,
check_name_availability_input: Union[_models.CheckNameAvailabilityInput, IO[bytes]],
**kwargs: Any
) -> _models.CheckNameAvailabilityOutput:
"""Check the availability of a resource name. This API should be used to check the uniqueness of
the name for adding a new communication to the support ticket.
:param support_ticket_name: Support ticket name. Required.
:type support_ticket_name: str
:param check_name_availability_input: Input to check. Is either a CheckNameAvailabilityInput
type or a IO[bytes] type. Required.
:type check_name_availability_input: ~azure.mgmt.support.models.CheckNameAvailabilityInput or
IO[bytes]
:return: CheckNameAvailabilityOutput or the result of cls(response)
:rtype: ~azure.mgmt.support.models.CheckNameAvailabilityOutput
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._config.api_version))
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
cls: ClsType[_models.CheckNameAvailabilityOutput] = kwargs.pop("cls", None)
content_type = content_type or "application/json"
_json = None
_content = None
if isinstance(check_name_availability_input, (IOBase, bytes)):
_content = check_name_availability_input
else:
_json = self._serialize.body(check_name_availability_input, "CheckNameAvailabilityInput")
_request = build_check_name_availability_request(
support_ticket_name=support_ticket_name,
subscription_id=self._config.subscription_id,
api_version=api_version,
content_type=content_type,
json=_json,
content=_content,
headers=_headers,
params=_params,
)
_request = _convert_request(_request)
_request.url = self._client.format_url(_request.url)
_stream = False
pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
_request, stream=_stream, **kwargs
)
response = pipeline_response.http_response
if response.status_code not in [200]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response)
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
deserialized = self._deserialize("CheckNameAvailabilityOutput", pipeline_response)
if cls:
return cls(pipeline_response, deserialized, {}) # type: ignore
return deserialized # type: ignore
[docs] @distributed_trace
def list(
self, support_ticket_name: str, top: Optional[int] = None, filter: Optional[str] = None, **kwargs: Any
) -> Iterable["_models.CommunicationDetails"]:
"""Lists all communications (attachments not included) for a support ticket. :code:`<br/>`</br>
You can also filter support ticket communications by *CreatedDate* or *CommunicationType* using
the $filter parameter. The only type of communication supported today is *Web*. Output will be
a paged result with *nextLink*\ , using which you can retrieve the next set of Communication
results. :code:`<br/>`:code:`<br/>`Support ticket data is available for 18 months after ticket
creation. If a ticket was created more than 18 months ago, a request for data might cause an
error.
:param support_ticket_name: Support ticket name. Required.
:type support_ticket_name: str
:param top: The number of values to return in the collection. Default is 10 and max is 10.
Default value is None.
:type top: int
:param filter: The filter to apply on the operation. You can filter by communicationType and
createdDate properties. CommunicationType supports Equals ('eq') operator and createdDate
supports Greater Than ('gt') and Greater Than or Equals ('ge') operators. You may combine the
CommunicationType and CreatedDate filters by Logical And ('and') operator. Default value is
None.
:type filter: str
:return: An iterator like instance of either CommunicationDetails or the result of
cls(response)
:rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.support.models.CommunicationDetails]
:raises ~azure.core.exceptions.HttpResponseError:
"""
_headers = kwargs.pop("headers", {}) or {}
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._config.api_version))
cls: ClsType[_models.CommunicationsListResult] = kwargs.pop("cls", None)
error_map = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
def prepare_request(next_link=None):
if not next_link:
_request = build_list_request(
support_ticket_name=support_ticket_name,
subscription_id=self._config.subscription_id,
top=top,
filter=filter,
api_version=api_version,
headers=_headers,
params=_params,
)
_request = _convert_request(_request)
_request.url = self._client.format_url(_request.url)
else:
# make call to next link with the client's api-version
_parsed_next_link = urllib.parse.urlparse(next_link)
_next_request_params = case_insensitive_dict(
{
key: [urllib.parse.quote(v) for v in value]
for key, value in urllib.parse.parse_qs(_parsed_next_link.query).items()
}
)
_next_request_params["api-version"] = self._config.api_version
_request = HttpRequest(
"GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params
)
_request = _convert_request(_request)
_request.url = self._client.format_url(_request.url)
_request.method = "GET"
return _request
def extract_data(pipeline_response):
deserialized = self._deserialize("CommunicationsListResult", pipeline_response)
list_of_elem = deserialized.value
if cls:
list_of_elem = cls(list_of_elem) # type: ignore
return deserialized.next_link or None, iter(list_of_elem)
def get_next(next_link=None):
_request = prepare_request(next_link)
_stream = False
pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
_request, stream=_stream, **kwargs
)
response = pipeline_response.http_response
if response.status_code not in [200]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response)
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
return pipeline_response
return ItemPaged(get_next, extract_data)
[docs] @distributed_trace
def get(self, support_ticket_name: str, communication_name: str, **kwargs: Any) -> _models.CommunicationDetails:
"""Returns communication details for a support ticket.
:param support_ticket_name: Support ticket name. Required.
:type support_ticket_name: str
:param communication_name: Communication name. Required.
:type communication_name: str
:return: CommunicationDetails or the result of cls(response)
:rtype: ~azure.mgmt.support.models.CommunicationDetails
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = kwargs.pop("headers", {}) or {}
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._config.api_version))
cls: ClsType[_models.CommunicationDetails] = kwargs.pop("cls", None)
_request = build_get_request(
support_ticket_name=support_ticket_name,
communication_name=communication_name,
subscription_id=self._config.subscription_id,
api_version=api_version,
headers=_headers,
params=_params,
)
_request = _convert_request(_request)
_request.url = self._client.format_url(_request.url)
_stream = False
pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
_request, stream=_stream, **kwargs
)
response = pipeline_response.http_response
if response.status_code not in [200]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response)
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
deserialized = self._deserialize("CommunicationDetails", pipeline_response)
if cls:
return cls(pipeline_response, deserialized, {}) # type: ignore
return deserialized # type: ignore
def _create_initial(
self,
support_ticket_name: str,
communication_name: str,
create_communication_parameters: Union[_models.CommunicationDetails, IO[bytes]],
**kwargs: Any
) -> Optional[_models.CommunicationDetails]:
error_map = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._config.api_version))
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
cls: ClsType[Optional[_models.CommunicationDetails]] = kwargs.pop("cls", None)
content_type = content_type or "application/json"
_json = None
_content = None
if isinstance(create_communication_parameters, (IOBase, bytes)):
_content = create_communication_parameters
else:
_json = self._serialize.body(create_communication_parameters, "CommunicationDetails")
_request = build_create_request(
support_ticket_name=support_ticket_name,
communication_name=communication_name,
subscription_id=self._config.subscription_id,
api_version=api_version,
content_type=content_type,
json=_json,
content=_content,
headers=_headers,
params=_params,
)
_request = _convert_request(_request)
_request.url = self._client.format_url(_request.url)
_stream = False
pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
_request, stream=_stream, **kwargs
)
response = pipeline_response.http_response
if response.status_code not in [200, 202]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response)
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
deserialized = None
if response.status_code == 200:
deserialized = self._deserialize("CommunicationDetails", pipeline_response)
if cls:
return cls(pipeline_response, deserialized, {}) # type: ignore
return deserialized # type: ignore
@overload
def begin_create(
self,
support_ticket_name: str,
communication_name: str,
create_communication_parameters: _models.CommunicationDetails,
*,
content_type: str = "application/json",
**kwargs: Any
) -> LROPoller[_models.CommunicationDetails]:
"""Adds a new customer communication to an Azure support ticket.
:param support_ticket_name: Support ticket name. Required.
:type support_ticket_name: str
:param communication_name: Communication name. Required.
:type communication_name: str
:param create_communication_parameters: Communication object. Required.
:type create_communication_parameters: ~azure.mgmt.support.models.CommunicationDetails
:keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
Default value is "application/json".
:paramtype content_type: str
:return: An instance of LROPoller that returns either CommunicationDetails or the result of
cls(response)
:rtype: ~azure.core.polling.LROPoller[~azure.mgmt.support.models.CommunicationDetails]
:raises ~azure.core.exceptions.HttpResponseError:
"""
@overload
def begin_create(
self,
support_ticket_name: str,
communication_name: str,
create_communication_parameters: IO[bytes],
*,
content_type: str = "application/json",
**kwargs: Any
) -> LROPoller[_models.CommunicationDetails]:
"""Adds a new customer communication to an Azure support ticket.
:param support_ticket_name: Support ticket name. Required.
:type support_ticket_name: str
:param communication_name: Communication name. Required.
:type communication_name: str
:param create_communication_parameters: Communication object. Required.
:type create_communication_parameters: IO[bytes]
:keyword content_type: Body Parameter content-type. Content type parameter for binary body.
Default value is "application/json".
:paramtype content_type: str
:return: An instance of LROPoller that returns either CommunicationDetails or the result of
cls(response)
:rtype: ~azure.core.polling.LROPoller[~azure.mgmt.support.models.CommunicationDetails]
:raises ~azure.core.exceptions.HttpResponseError:
"""
[docs] @distributed_trace
def begin_create(
self,
support_ticket_name: str,
communication_name: str,
create_communication_parameters: Union[_models.CommunicationDetails, IO[bytes]],
**kwargs: Any
) -> LROPoller[_models.CommunicationDetails]:
"""Adds a new customer communication to an Azure support ticket.
:param support_ticket_name: Support ticket name. Required.
:type support_ticket_name: str
:param communication_name: Communication name. Required.
:type communication_name: str
:param create_communication_parameters: Communication object. Is either a CommunicationDetails
type or a IO[bytes] type. Required.
:type create_communication_parameters: ~azure.mgmt.support.models.CommunicationDetails or
IO[bytes]
:return: An instance of LROPoller that returns either CommunicationDetails or the result of
cls(response)
:rtype: ~azure.core.polling.LROPoller[~azure.mgmt.support.models.CommunicationDetails]
:raises ~azure.core.exceptions.HttpResponseError:
"""
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._config.api_version))
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
cls: ClsType[_models.CommunicationDetails] = kwargs.pop("cls", None)
polling: Union[bool, PollingMethod] = kwargs.pop("polling", True)
lro_delay = kwargs.pop("polling_interval", self._config.polling_interval)
cont_token: Optional[str] = kwargs.pop("continuation_token", None)
if cont_token is None:
raw_result = self._create_initial(
support_ticket_name=support_ticket_name,
communication_name=communication_name,
create_communication_parameters=create_communication_parameters,
api_version=api_version,
content_type=content_type,
cls=lambda x, y, z: x,
headers=_headers,
params=_params,
**kwargs
)
kwargs.pop("error_map", None)
def get_long_running_output(pipeline_response):
deserialized = self._deserialize("CommunicationDetails", pipeline_response)
if cls:
return cls(pipeline_response, deserialized, {}) # type: ignore
return deserialized
if polling is True:
polling_method: PollingMethod = cast(
PollingMethod, ARMPolling(lro_delay, lro_options={"final-state-via": "azure-async-operation"}, **kwargs)
)
elif polling is False:
polling_method = cast(PollingMethod, NoPolling())
else:
polling_method = polling
if cont_token:
return LROPoller[_models.CommunicationDetails].from_continuation_token(
polling_method=polling_method,
continuation_token=cont_token,
client=self._client,
deserialization_callback=get_long_running_output,
)
return LROPoller[_models.CommunicationDetails](
self._client, raw_result, get_long_running_output, polling_method # type: ignore
)