# pylint: disable=too-many-lines
# 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.
# --------------------------------------------------------------------------
import sys
from typing import Any, Callable, Dict, IO, Iterable, Optional, TypeVar, Union, 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.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 .. import models as _models
from .._serialization import Serializer
from .._vendor import _convert_request, _format_url_section
if sys.version_info >= (3, 8):
from typing import Literal # pylint: disable=no-name-in-module, ungrouped-imports
else:
from typing_extensions import Literal # type: ignore # pylint: disable=ungrouped-imports
T = TypeVar("T")
ClsType = Optional[Callable[[PipelineResponse[HttpRequest, HttpResponse], T, Dict[str, Any]], Any]]
_SERIALIZER = Serializer()
_SERIALIZER.client_side_validation = False
def build_create_or_update_request(
resource_group_name: str,
private_zone_name: str,
record_type: Union[str, _models.RecordType],
relative_record_set_name: str,
subscription_id: str,
*,
if_match: Optional[str] = None,
if_none_match: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
api_version = kwargs.pop("api_version", _params.pop("api-version", "2020-06-01")) # type: Literal["2020-06-01"]
content_type = kwargs.pop("content_type", _headers.pop("Content-Type", None)) # type: Optional[str]
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = kwargs.pop(
"template_url",
"/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateDnsZones/{privateZoneName}/{recordType}/{relativeRecordSetName}",
) # pylint: disable=line-too-long
path_format_arguments = {
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
"privateZoneName": _SERIALIZER.url("private_zone_name", private_zone_name, "str"),
"recordType": _SERIALIZER.url("record_type", record_type, "str"),
"relativeRecordSetName": _SERIALIZER.url(
"relative_record_set_name", relative_record_set_name, "str", skip_quote=True
),
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
}
_url = _format_url_section(_url, **path_format_arguments)
# Construct parameters
_params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
# Construct headers
if if_match is not None:
_headers["If-Match"] = _SERIALIZER.header("if_match", if_match, "str")
if if_none_match is not None:
_headers["If-None-Match"] = _SERIALIZER.header("if_none_match", if_none_match, "str")
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)
def build_update_request(
resource_group_name: str,
private_zone_name: str,
record_type: Union[str, _models.RecordType],
relative_record_set_name: str,
subscription_id: str,
*,
if_match: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
api_version = kwargs.pop("api_version", _params.pop("api-version", "2020-06-01")) # type: Literal["2020-06-01"]
content_type = kwargs.pop("content_type", _headers.pop("Content-Type", None)) # type: Optional[str]
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = kwargs.pop(
"template_url",
"/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateDnsZones/{privateZoneName}/{recordType}/{relativeRecordSetName}",
) # pylint: disable=line-too-long
path_format_arguments = {
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
"privateZoneName": _SERIALIZER.url("private_zone_name", private_zone_name, "str"),
"recordType": _SERIALIZER.url("record_type", record_type, "str"),
"relativeRecordSetName": _SERIALIZER.url(
"relative_record_set_name", relative_record_set_name, "str", skip_quote=True
),
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
}
_url = _format_url_section(_url, **path_format_arguments)
# Construct parameters
_params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
# Construct headers
if if_match is not None:
_headers["If-Match"] = _SERIALIZER.header("if_match", if_match, "str")
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="PATCH", url=_url, params=_params, headers=_headers, **kwargs)
def build_delete_request(
resource_group_name: str,
private_zone_name: str,
record_type: Union[str, _models.RecordType],
relative_record_set_name: str,
subscription_id: str,
*,
if_match: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
api_version = kwargs.pop("api_version", _params.pop("api-version", "2020-06-01")) # type: Literal["2020-06-01"]
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = kwargs.pop(
"template_url",
"/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateDnsZones/{privateZoneName}/{recordType}/{relativeRecordSetName}",
) # pylint: disable=line-too-long
path_format_arguments = {
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
"privateZoneName": _SERIALIZER.url("private_zone_name", private_zone_name, "str"),
"recordType": _SERIALIZER.url("record_type", record_type, "str"),
"relativeRecordSetName": _SERIALIZER.url(
"relative_record_set_name", relative_record_set_name, "str", skip_quote=True
),
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
}
_url = _format_url_section(_url, **path_format_arguments)
# Construct parameters
_params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
# Construct headers
if if_match is not None:
_headers["If-Match"] = _SERIALIZER.header("if_match", if_match, "str")
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="DELETE", url=_url, params=_params, headers=_headers, **kwargs)
def build_get_request(
resource_group_name: str,
private_zone_name: str,
record_type: Union[str, _models.RecordType],
relative_record_set_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 = kwargs.pop("api_version", _params.pop("api-version", "2020-06-01")) # type: Literal["2020-06-01"]
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = kwargs.pop(
"template_url",
"/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateDnsZones/{privateZoneName}/{recordType}/{relativeRecordSetName}",
) # pylint: disable=line-too-long
path_format_arguments = {
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
"privateZoneName": _SERIALIZER.url("private_zone_name", private_zone_name, "str"),
"recordType": _SERIALIZER.url("record_type", record_type, "str"),
"relativeRecordSetName": _SERIALIZER.url(
"relative_record_set_name", relative_record_set_name, "str", skip_quote=True
),
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
}
_url = _format_url_section(_url, **path_format_arguments)
# 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_list_by_type_request(
resource_group_name: str,
private_zone_name: str,
record_type: Union[str, _models.RecordType],
subscription_id: str,
*,
top: Optional[int] = None,
recordsetnamesuffix: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
api_version = kwargs.pop("api_version", _params.pop("api-version", "2020-06-01")) # type: Literal["2020-06-01"]
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = kwargs.pop(
"template_url",
"/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateDnsZones/{privateZoneName}/{recordType}",
) # pylint: disable=line-too-long
path_format_arguments = {
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
"privateZoneName": _SERIALIZER.url("private_zone_name", private_zone_name, "str"),
"recordType": _SERIALIZER.url("record_type", record_type, "str"),
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
}
_url = _format_url_section(_url, **path_format_arguments)
# Construct parameters
if top is not None:
_params["$top"] = _SERIALIZER.query("top", top, "int")
if recordsetnamesuffix is not None:
_params["$recordsetnamesuffix"] = _SERIALIZER.query("recordsetnamesuffix", recordsetnamesuffix, "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_list_request(
resource_group_name: str,
private_zone_name: str,
subscription_id: str,
*,
top: Optional[int] = None,
recordsetnamesuffix: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
api_version = kwargs.pop("api_version", _params.pop("api-version", "2020-06-01")) # type: Literal["2020-06-01"]
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = kwargs.pop(
"template_url",
"/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateDnsZones/{privateZoneName}/ALL",
) # pylint: disable=line-too-long
path_format_arguments = {
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
"privateZoneName": _SERIALIZER.url("private_zone_name", private_zone_name, "str"),
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
}
_url = _format_url_section(_url, **path_format_arguments)
# Construct parameters
if top is not None:
_params["$top"] = _SERIALIZER.query("top", top, "int")
if recordsetnamesuffix is not None:
_params["$recordsetnamesuffix"] = _SERIALIZER.query("recordsetnamesuffix", recordsetnamesuffix, "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)
[docs]class RecordSetsOperations:
"""
.. warning::
**DO NOT** instantiate this class directly.
Instead, you should access the following operations through
:class:`~azure.mgmt.privatedns.PrivateDnsManagementClient`'s
:attr:`record_sets` 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 create_or_update(
self,
resource_group_name: str,
private_zone_name: str,
record_type: Union[str, _models.RecordType],
relative_record_set_name: str,
parameters: _models.RecordSet,
if_match: Optional[str] = None,
if_none_match: Optional[str] = None,
*,
content_type: str = "application/json",
**kwargs: Any
) -> _models.RecordSet:
"""Creates or updates a record set within a Private DNS zone.
:param resource_group_name: The name of the resource group. Required.
:type resource_group_name: str
:param private_zone_name: The name of the Private DNS zone (without a terminating dot).
Required.
:type private_zone_name: str
:param record_type: The type of DNS record in this record set. Record sets of type SOA can be
updated but not created (they are created when the Private DNS zone is created). Known values
are: "A", "AAAA", "CNAME", "MX", "PTR", "SOA", "SRV", and "TXT". Required.
:type record_type: str or ~azure.mgmt.privatedns.models.RecordType
:param relative_record_set_name: The name of the record set, relative to the name of the zone.
Required.
:type relative_record_set_name: str
:param parameters: Parameters supplied to the CreateOrUpdate operation. Required.
:type parameters: ~azure.mgmt.privatedns.models.RecordSet
:param if_match: The ETag of the record set. Omit this value to always overwrite the current
record set. Specify the last-seen ETag value to prevent accidentally overwriting any concurrent
changes. Default value is None.
:type if_match: str
:param if_none_match: Set to '*' to allow a new record set to be created, but to prevent
updating an existing record set. Other values will be ignored. Default value is None.
:type if_none_match: str
:keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
Default value is "application/json".
:paramtype content_type: str
:keyword callable cls: A custom type or function that will be passed the direct response
:return: RecordSet or the result of cls(response)
:rtype: ~azure.mgmt.privatedns.models.RecordSet
:raises ~azure.core.exceptions.HttpResponseError:
"""
@overload
def create_or_update(
self,
resource_group_name: str,
private_zone_name: str,
record_type: Union[str, _models.RecordType],
relative_record_set_name: str,
parameters: IO,
if_match: Optional[str] = None,
if_none_match: Optional[str] = None,
*,
content_type: str = "application/json",
**kwargs: Any
) -> _models.RecordSet:
"""Creates or updates a record set within a Private DNS zone.
:param resource_group_name: The name of the resource group. Required.
:type resource_group_name: str
:param private_zone_name: The name of the Private DNS zone (without a terminating dot).
Required.
:type private_zone_name: str
:param record_type: The type of DNS record in this record set. Record sets of type SOA can be
updated but not created (they are created when the Private DNS zone is created). Known values
are: "A", "AAAA", "CNAME", "MX", "PTR", "SOA", "SRV", and "TXT". Required.
:type record_type: str or ~azure.mgmt.privatedns.models.RecordType
:param relative_record_set_name: The name of the record set, relative to the name of the zone.
Required.
:type relative_record_set_name: str
:param parameters: Parameters supplied to the CreateOrUpdate operation. Required.
:type parameters: IO
:param if_match: The ETag of the record set. Omit this value to always overwrite the current
record set. Specify the last-seen ETag value to prevent accidentally overwriting any concurrent
changes. Default value is None.
:type if_match: str
:param if_none_match: Set to '*' to allow a new record set to be created, but to prevent
updating an existing record set. Other values will be ignored. Default value is None.
:type if_none_match: str
:keyword content_type: Body Parameter content-type. Content type parameter for binary body.
Default value is "application/json".
:paramtype content_type: str
:keyword callable cls: A custom type or function that will be passed the direct response
:return: RecordSet or the result of cls(response)
:rtype: ~azure.mgmt.privatedns.models.RecordSet
:raises ~azure.core.exceptions.HttpResponseError:
"""
[docs] @distributed_trace
def create_or_update(
self,
resource_group_name: str,
private_zone_name: str,
record_type: Union[str, _models.RecordType],
relative_record_set_name: str,
parameters: Union[_models.RecordSet, IO],
if_match: Optional[str] = None,
if_none_match: Optional[str] = None,
**kwargs: Any
) -> _models.RecordSet:
"""Creates or updates a record set within a Private DNS zone.
:param resource_group_name: The name of the resource group. Required.
:type resource_group_name: str
:param private_zone_name: The name of the Private DNS zone (without a terminating dot).
Required.
:type private_zone_name: str
:param record_type: The type of DNS record in this record set. Record sets of type SOA can be
updated but not created (they are created when the Private DNS zone is created). Known values
are: "A", "AAAA", "CNAME", "MX", "PTR", "SOA", "SRV", and "TXT". Required.
:type record_type: str or ~azure.mgmt.privatedns.models.RecordType
:param relative_record_set_name: The name of the record set, relative to the name of the zone.
Required.
:type relative_record_set_name: str
:param parameters: Parameters supplied to the CreateOrUpdate operation. Is either a model type
or a IO type. Required.
:type parameters: ~azure.mgmt.privatedns.models.RecordSet or IO
:param if_match: The ETag of the record set. Omit this value to always overwrite the current
record set. Specify the last-seen ETag value to prevent accidentally overwriting any concurrent
changes. Default value is None.
:type if_match: str
:param if_none_match: Set to '*' to allow a new record set to be created, but to prevent
updating an existing record set. Other values will be ignored. Default value is None.
:type if_none_match: str
:keyword content_type: Body Parameter content-type. Known values are: 'application/json'.
Default value is None.
:paramtype content_type: str
:keyword callable cls: A custom type or function that will be passed the direct response
:return: RecordSet or the result of cls(response)
:rtype: ~azure.mgmt.privatedns.models.RecordSet
: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 = kwargs.pop(
"api_version", _params.pop("api-version", self._config.api_version)
) # type: Literal["2020-06-01"]
content_type = kwargs.pop("content_type", _headers.pop("Content-Type", None)) # type: Optional[str]
cls = kwargs.pop("cls", None) # type: ClsType[_models.RecordSet]
content_type = content_type or "application/json"
_json = None
_content = None
if isinstance(parameters, (IO, bytes)):
_content = parameters
else:
_json = self._serialize.body(parameters, "RecordSet")
request = build_create_or_update_request(
resource_group_name=resource_group_name,
private_zone_name=private_zone_name,
record_type=record_type,
relative_record_set_name=relative_record_set_name,
subscription_id=self._config.subscription_id,
if_match=if_match,
if_none_match=if_none_match,
api_version=api_version,
content_type=content_type,
json=_json,
content=_content,
template_url=self.create_or_update.metadata["url"],
headers=_headers,
params=_params,
)
request = _convert_request(request)
request.url = self._client.format_url(request.url) # type: ignore
pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access
request, stream=False, **kwargs
)
response = pipeline_response.http_response
if response.status_code not in [200, 201]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
raise HttpResponseError(response=response, error_format=ARMErrorFormat)
if response.status_code == 200:
deserialized = self._deserialize("RecordSet", pipeline_response)
if response.status_code == 201:
deserialized = self._deserialize("RecordSet", pipeline_response)
if cls:
return cls(pipeline_response, deserialized, {})
return deserialized
create_or_update.metadata = {"url": "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateDnsZones/{privateZoneName}/{recordType}/{relativeRecordSetName}"} # type: ignore
@overload
def update(
self,
resource_group_name: str,
private_zone_name: str,
record_type: Union[str, _models.RecordType],
relative_record_set_name: str,
parameters: _models.RecordSet,
if_match: Optional[str] = None,
*,
content_type: str = "application/json",
**kwargs: Any
) -> _models.RecordSet:
"""Updates a record set within a Private DNS zone.
:param resource_group_name: The name of the resource group. Required.
:type resource_group_name: str
:param private_zone_name: The name of the Private DNS zone (without a terminating dot).
Required.
:type private_zone_name: str
:param record_type: The type of DNS record in this record set. Known values are: "A", "AAAA",
"CNAME", "MX", "PTR", "SOA", "SRV", and "TXT". Required.
:type record_type: str or ~azure.mgmt.privatedns.models.RecordType
:param relative_record_set_name: The name of the record set, relative to the name of the zone.
Required.
:type relative_record_set_name: str
:param parameters: Parameters supplied to the Update operation. Required.
:type parameters: ~azure.mgmt.privatedns.models.RecordSet
:param if_match: The ETag of the record set. Omit this value to always overwrite the current
record set. Specify the last-seen ETag value to prevent accidentally overwriting concurrent
changes. Default value is None.
:type if_match: str
:keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
Default value is "application/json".
:paramtype content_type: str
:keyword callable cls: A custom type or function that will be passed the direct response
:return: RecordSet or the result of cls(response)
:rtype: ~azure.mgmt.privatedns.models.RecordSet
:raises ~azure.core.exceptions.HttpResponseError:
"""
@overload
def update(
self,
resource_group_name: str,
private_zone_name: str,
record_type: Union[str, _models.RecordType],
relative_record_set_name: str,
parameters: IO,
if_match: Optional[str] = None,
*,
content_type: str = "application/json",
**kwargs: Any
) -> _models.RecordSet:
"""Updates a record set within a Private DNS zone.
:param resource_group_name: The name of the resource group. Required.
:type resource_group_name: str
:param private_zone_name: The name of the Private DNS zone (without a terminating dot).
Required.
:type private_zone_name: str
:param record_type: The type of DNS record in this record set. Known values are: "A", "AAAA",
"CNAME", "MX", "PTR", "SOA", "SRV", and "TXT". Required.
:type record_type: str or ~azure.mgmt.privatedns.models.RecordType
:param relative_record_set_name: The name of the record set, relative to the name of the zone.
Required.
:type relative_record_set_name: str
:param parameters: Parameters supplied to the Update operation. Required.
:type parameters: IO
:param if_match: The ETag of the record set. Omit this value to always overwrite the current
record set. Specify the last-seen ETag value to prevent accidentally overwriting concurrent
changes. Default value is None.
:type if_match: str
:keyword content_type: Body Parameter content-type. Content type parameter for binary body.
Default value is "application/json".
:paramtype content_type: str
:keyword callable cls: A custom type or function that will be passed the direct response
:return: RecordSet or the result of cls(response)
:rtype: ~azure.mgmt.privatedns.models.RecordSet
:raises ~azure.core.exceptions.HttpResponseError:
"""
[docs] @distributed_trace
def update(
self,
resource_group_name: str,
private_zone_name: str,
record_type: Union[str, _models.RecordType],
relative_record_set_name: str,
parameters: Union[_models.RecordSet, IO],
if_match: Optional[str] = None,
**kwargs: Any
) -> _models.RecordSet:
"""Updates a record set within a Private DNS zone.
:param resource_group_name: The name of the resource group. Required.
:type resource_group_name: str
:param private_zone_name: The name of the Private DNS zone (without a terminating dot).
Required.
:type private_zone_name: str
:param record_type: The type of DNS record in this record set. Known values are: "A", "AAAA",
"CNAME", "MX", "PTR", "SOA", "SRV", and "TXT". Required.
:type record_type: str or ~azure.mgmt.privatedns.models.RecordType
:param relative_record_set_name: The name of the record set, relative to the name of the zone.
Required.
:type relative_record_set_name: str
:param parameters: Parameters supplied to the Update operation. Is either a model type or a IO
type. Required.
:type parameters: ~azure.mgmt.privatedns.models.RecordSet or IO
:param if_match: The ETag of the record set. Omit this value to always overwrite the current
record set. Specify the last-seen ETag value to prevent accidentally overwriting concurrent
changes. Default value is None.
:type if_match: str
:keyword content_type: Body Parameter content-type. Known values are: 'application/json'.
Default value is None.
:paramtype content_type: str
:keyword callable cls: A custom type or function that will be passed the direct response
:return: RecordSet or the result of cls(response)
:rtype: ~azure.mgmt.privatedns.models.RecordSet
: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 = kwargs.pop(
"api_version", _params.pop("api-version", self._config.api_version)
) # type: Literal["2020-06-01"]
content_type = kwargs.pop("content_type", _headers.pop("Content-Type", None)) # type: Optional[str]
cls = kwargs.pop("cls", None) # type: ClsType[_models.RecordSet]
content_type = content_type or "application/json"
_json = None
_content = None
if isinstance(parameters, (IO, bytes)):
_content = parameters
else:
_json = self._serialize.body(parameters, "RecordSet")
request = build_update_request(
resource_group_name=resource_group_name,
private_zone_name=private_zone_name,
record_type=record_type,
relative_record_set_name=relative_record_set_name,
subscription_id=self._config.subscription_id,
if_match=if_match,
api_version=api_version,
content_type=content_type,
json=_json,
content=_content,
template_url=self.update.metadata["url"],
headers=_headers,
params=_params,
)
request = _convert_request(request)
request.url = self._client.format_url(request.url) # type: ignore
pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access
request, stream=False, **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)
raise HttpResponseError(response=response, error_format=ARMErrorFormat)
deserialized = self._deserialize("RecordSet", pipeline_response)
if cls:
return cls(pipeline_response, deserialized, {})
return deserialized
update.metadata = {"url": "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateDnsZones/{privateZoneName}/{recordType}/{relativeRecordSetName}"} # type: ignore
[docs] @distributed_trace
def delete( # pylint: disable=inconsistent-return-statements
self,
resource_group_name: str,
private_zone_name: str,
record_type: Union[str, _models.RecordType],
relative_record_set_name: str,
if_match: Optional[str] = None,
**kwargs: Any
) -> None:
"""Deletes a record set from a Private DNS zone. This operation cannot be undone.
:param resource_group_name: The name of the resource group. Required.
:type resource_group_name: str
:param private_zone_name: The name of the Private DNS zone (without a terminating dot).
Required.
:type private_zone_name: str
:param record_type: The type of DNS record in this record set. Record sets of type SOA cannot
be deleted (they are deleted when the Private DNS zone is deleted). Known values are: "A",
"AAAA", "CNAME", "MX", "PTR", "SOA", "SRV", and "TXT". Required.
:type record_type: str or ~azure.mgmt.privatedns.models.RecordType
:param relative_record_set_name: The name of the record set, relative to the name of the zone.
Required.
:type relative_record_set_name: str
:param if_match: The ETag of the record set. Omit this value to always delete the current
record set. Specify the last-seen ETag value to prevent accidentally deleting any concurrent
changes. Default value is None.
:type if_match: str
:keyword callable cls: A custom type or function that will be passed the direct response
:return: None or the result of cls(response)
:rtype: None
: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 = kwargs.pop(
"api_version", _params.pop("api-version", self._config.api_version)
) # type: Literal["2020-06-01"]
cls = kwargs.pop("cls", None) # type: ClsType[None]
request = build_delete_request(
resource_group_name=resource_group_name,
private_zone_name=private_zone_name,
record_type=record_type,
relative_record_set_name=relative_record_set_name,
subscription_id=self._config.subscription_id,
if_match=if_match,
api_version=api_version,
template_url=self.delete.metadata["url"],
headers=_headers,
params=_params,
)
request = _convert_request(request)
request.url = self._client.format_url(request.url) # type: ignore
pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access
request, stream=False, **kwargs
)
response = pipeline_response.http_response
if response.status_code not in [200, 204]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
raise HttpResponseError(response=response, error_format=ARMErrorFormat)
if cls:
return cls(pipeline_response, None, {})
delete.metadata = {"url": "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateDnsZones/{privateZoneName}/{recordType}/{relativeRecordSetName}"} # type: ignore
[docs] @distributed_trace
def get(
self,
resource_group_name: str,
private_zone_name: str,
record_type: Union[str, _models.RecordType],
relative_record_set_name: str,
**kwargs: Any
) -> _models.RecordSet:
"""Gets a record set.
:param resource_group_name: The name of the resource group. Required.
:type resource_group_name: str
:param private_zone_name: The name of the Private DNS zone (without a terminating dot).
Required.
:type private_zone_name: str
:param record_type: The type of DNS record in this record set. Known values are: "A", "AAAA",
"CNAME", "MX", "PTR", "SOA", "SRV", and "TXT". Required.
:type record_type: str or ~azure.mgmt.privatedns.models.RecordType
:param relative_record_set_name: The name of the record set, relative to the name of the zone.
Required.
:type relative_record_set_name: str
:keyword callable cls: A custom type or function that will be passed the direct response
:return: RecordSet or the result of cls(response)
:rtype: ~azure.mgmt.privatedns.models.RecordSet
: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 = kwargs.pop(
"api_version", _params.pop("api-version", self._config.api_version)
) # type: Literal["2020-06-01"]
cls = kwargs.pop("cls", None) # type: ClsType[_models.RecordSet]
request = build_get_request(
resource_group_name=resource_group_name,
private_zone_name=private_zone_name,
record_type=record_type,
relative_record_set_name=relative_record_set_name,
subscription_id=self._config.subscription_id,
api_version=api_version,
template_url=self.get.metadata["url"],
headers=_headers,
params=_params,
)
request = _convert_request(request)
request.url = self._client.format_url(request.url) # type: ignore
pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access
request, stream=False, **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)
raise HttpResponseError(response=response, error_format=ARMErrorFormat)
deserialized = self._deserialize("RecordSet", pipeline_response)
if cls:
return cls(pipeline_response, deserialized, {})
return deserialized
get.metadata = {"url": "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateDnsZones/{privateZoneName}/{recordType}/{relativeRecordSetName}"} # type: ignore
[docs] @distributed_trace
def list_by_type(
self,
resource_group_name: str,
private_zone_name: str,
record_type: Union[str, _models.RecordType],
top: Optional[int] = None,
recordsetnamesuffix: Optional[str] = None,
**kwargs: Any
) -> Iterable["_models.RecordSet"]:
"""Lists the record sets of a specified type in a Private DNS zone.
:param resource_group_name: The name of the resource group. Required.
:type resource_group_name: str
:param private_zone_name: The name of the Private DNS zone (without a terminating dot).
Required.
:type private_zone_name: str
:param record_type: The type of record sets to enumerate. Known values are: "A", "AAAA",
"CNAME", "MX", "PTR", "SOA", "SRV", and "TXT". Required.
:type record_type: str or ~azure.mgmt.privatedns.models.RecordType
:param top: The maximum number of record sets to return. If not specified, returns up to 100
record sets. Default value is None.
:type top: int
:param recordsetnamesuffix: The suffix label of the record set name to be used to filter the
record set enumeration. If this parameter is specified, the returned enumeration will only
contain records that end with ".:code:`<recordsetnamesuffix>`". Default value is None.
:type recordsetnamesuffix: str
:keyword callable cls: A custom type or function that will be passed the direct response
:return: An iterator like instance of either RecordSet or the result of cls(response)
:rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.privatedns.models.RecordSet]
:raises ~azure.core.exceptions.HttpResponseError:
"""
_headers = kwargs.pop("headers", {}) or {}
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
api_version = kwargs.pop(
"api_version", _params.pop("api-version", self._config.api_version)
) # type: Literal["2020-06-01"]
cls = kwargs.pop("cls", None) # type: ClsType[_models.RecordSetListResult]
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_by_type_request(
resource_group_name=resource_group_name,
private_zone_name=private_zone_name,
record_type=record_type,
subscription_id=self._config.subscription_id,
top=top,
recordsetnamesuffix=recordsetnamesuffix,
api_version=api_version,
template_url=self.list_by_type.metadata["url"],
headers=_headers,
params=_params,
)
request = _convert_request(request)
request.url = self._client.format_url(request.url) # type: ignore
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) # type: ignore
request.method = "GET"
return request
def extract_data(pipeline_response):
deserialized = self._deserialize("RecordSetListResult", pipeline_response)
list_of_elem = deserialized.value
if cls:
list_of_elem = cls(list_of_elem)
return deserialized.next_link or None, iter(list_of_elem)
def get_next(next_link=None):
request = prepare_request(next_link)
pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access
request, stream=False, **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)
raise HttpResponseError(response=response, error_format=ARMErrorFormat)
return pipeline_response
return ItemPaged(get_next, extract_data)
list_by_type.metadata = {"url": "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateDnsZones/{privateZoneName}/{recordType}"} # type: ignore
[docs] @distributed_trace
def list(
self,
resource_group_name: str,
private_zone_name: str,
top: Optional[int] = None,
recordsetnamesuffix: Optional[str] = None,
**kwargs: Any
) -> Iterable["_models.RecordSet"]:
"""Lists all record sets in a Private DNS zone.
:param resource_group_name: The name of the resource group. Required.
:type resource_group_name: str
:param private_zone_name: The name of the Private DNS zone (without a terminating dot).
Required.
:type private_zone_name: str
:param top: The maximum number of record sets to return. If not specified, returns up to 100
record sets. Default value is None.
:type top: int
:param recordsetnamesuffix: The suffix label of the record set name to be used to filter the
record set enumeration. If this parameter is specified, the returned enumeration will only
contain records that end with ".:code:`<recordsetnamesuffix>`". Default value is None.
:type recordsetnamesuffix: str
:keyword callable cls: A custom type or function that will be passed the direct response
:return: An iterator like instance of either RecordSet or the result of cls(response)
:rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.privatedns.models.RecordSet]
:raises ~azure.core.exceptions.HttpResponseError:
"""
_headers = kwargs.pop("headers", {}) or {}
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
api_version = kwargs.pop(
"api_version", _params.pop("api-version", self._config.api_version)
) # type: Literal["2020-06-01"]
cls = kwargs.pop("cls", None) # type: ClsType[_models.RecordSetListResult]
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(
resource_group_name=resource_group_name,
private_zone_name=private_zone_name,
subscription_id=self._config.subscription_id,
top=top,
recordsetnamesuffix=recordsetnamesuffix,
api_version=api_version,
template_url=self.list.metadata["url"],
headers=_headers,
params=_params,
)
request = _convert_request(request)
request.url = self._client.format_url(request.url) # type: ignore
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) # type: ignore
request.method = "GET"
return request
def extract_data(pipeline_response):
deserialized = self._deserialize("RecordSetListResult", pipeline_response)
list_of_elem = deserialized.value
if cls:
list_of_elem = cls(list_of_elem)
return deserialized.next_link or None, iter(list_of_elem)
def get_next(next_link=None):
request = prepare_request(next_link)
pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access
request, stream=False, **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)
raise HttpResponseError(response=response, error_format=ARMErrorFormat)
return pipeline_response
return ItemPaged(get_next, extract_data)
list.metadata = {"url": "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateDnsZones/{privateZoneName}/ALL"} # type: ignore