# 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) Python Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is regenerated.
# --------------------------------------------------------------------------
from collections.abc import MutableMapping
from io import IOBase
import json
from typing import Any, Callable, Dict, IO, Iterable, Iterator, List, Optional, TypeVar, Union, cast, overload
import urllib.parse
from azure.core import PipelineClient
from azure.core.exceptions import (
ClientAuthenticationError,
HttpResponseError,
ResourceExistsError,
ResourceNotFoundError,
ResourceNotModifiedError,
StreamClosedError,
StreamConsumedError,
map_error,
)
from azure.core.paging import ItemPaged
from azure.core.pipeline import PipelineResponse
from azure.core.polling import LROPoller, NoPolling, PollingMethod
from azure.core.rest import HttpRequest, HttpResponse
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 .._configuration import SiteManagerMgmtClientConfiguration
from .._utils.model_base import SdkJSONEncoder, _deserialize, _failsafe_deserialize
from .._utils.serialization import Deserializer, Serializer
from .._validation import api_version_validation
JSON = MutableMapping[str, Any]
T = TypeVar("T")
ClsType = Optional[Callable[[PipelineResponse[HttpRequest, HttpResponse], T, Dict[str, Any]], Any]]
_SERIALIZER = Serializer()
_SERIALIZER.client_side_validation = False
def build_sites_get_request(
resource_group_name: str, site_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", "2025-03-01-preview"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = (
"/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Edge/sites/{siteName}"
)
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
"siteName": _SERIALIZER.url("site_name", site_name, "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_sites_create_or_update_request(
resource_group_name: str, site_name: str, subscription_id: str, **kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-03-01-preview"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = (
"/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Edge/sites/{siteName}"
)
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
"siteName": _SERIALIZER.url("site_name", site_name, "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)
def build_sites_update_request(
resource_group_name: str, site_name: str, subscription_id: str, **kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-03-01-preview"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = (
"/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Edge/sites/{siteName}"
)
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
"siteName": _SERIALIZER.url("site_name", site_name, "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="PATCH", url=_url, params=_params, headers=_headers, **kwargs)
def build_sites_delete_request(
resource_group_name: str, site_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", "2025-03-01-preview"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = (
"/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Edge/sites/{siteName}"
)
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
"siteName": _SERIALIZER.url("site_name", site_name, "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="DELETE", url=_url, params=_params, headers=_headers, **kwargs)
def build_sites_list_by_resource_group_request( # pylint: disable=name-too-long
resource_group_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", "2025-03-01-preview"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Edge/sites"
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "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_sites_by_subscription_list_request(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", "2025-03-01-preview"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/subscriptions/{subscriptionId}/providers/Microsoft.Edge/sites"
path_format_arguments = {
"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_sites_by_subscription_get_request(site_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", "2025-03-01-preview"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/subscriptions/{subscriptionId}/providers/Microsoft.Edge/sites/{siteName}"
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
"siteName": _SERIALIZER.url("site_name", site_name, "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_sites_by_subscription_create_or_update_request( # pylint: disable=name-too-long
site_name: str, subscription_id: str, **kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-03-01-preview"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/subscriptions/{subscriptionId}/providers/Microsoft.Edge/sites/{siteName}"
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
"siteName": _SERIALIZER.url("site_name", site_name, "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)
def build_sites_by_subscription_update_request( # pylint: disable=name-too-long
site_name: str, subscription_id: str, **kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-03-01-preview"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/subscriptions/{subscriptionId}/providers/Microsoft.Edge/sites/{siteName}"
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
"siteName": _SERIALIZER.url("site_name", site_name, "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="PATCH", url=_url, params=_params, headers=_headers, **kwargs)
def build_sites_by_subscription_delete_request( # pylint: disable=name-too-long
site_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", "2025-03-01-preview"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/subscriptions/{subscriptionId}/providers/Microsoft.Edge/sites/{siteName}"
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
"siteName": _SERIALIZER.url("site_name", site_name, "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="DELETE", url=_url, params=_params, headers=_headers, **kwargs)
def build_sites_by_service_group_list_by_service_group_request( # pylint: disable=name-too-long
servicegroup_name: 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", "2025-03-01-preview"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/providers/Microsoft.Management/serviceGroups/{servicegroupName}/providers/Microsoft.Edge/sites"
path_format_arguments = {
"servicegroupName": _SERIALIZER.url("servicegroup_name", servicegroup_name, "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_sites_by_service_group_get_request(servicegroup_name: str, site_name: 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", "2025-03-01-preview"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/providers/Microsoft.Management/serviceGroups/{servicegroupName}/providers/Microsoft.Edge/sites/{siteName}"
path_format_arguments = {
"servicegroupName": _SERIALIZER.url("servicegroup_name", servicegroup_name, "str"),
"siteName": _SERIALIZER.url("site_name", site_name, "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_sites_by_service_group_create_or_update_request( # pylint: disable=name-too-long
servicegroup_name: str, site_name: str, **kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-03-01-preview"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/providers/Microsoft.Management/serviceGroups/{servicegroupName}/providers/Microsoft.Edge/sites/{siteName}"
path_format_arguments = {
"servicegroupName": _SERIALIZER.url("servicegroup_name", servicegroup_name, "str"),
"siteName": _SERIALIZER.url("site_name", site_name, "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)
def build_sites_by_service_group_update_request( # pylint: disable=name-too-long
servicegroup_name: str, site_name: str, **kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-03-01-preview"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/providers/Microsoft.Management/serviceGroups/{servicegroupName}/providers/Microsoft.Edge/sites/{siteName}"
path_format_arguments = {
"servicegroupName": _SERIALIZER.url("servicegroup_name", servicegroup_name, "str"),
"siteName": _SERIALIZER.url("site_name", site_name, "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="PATCH", url=_url, params=_params, headers=_headers, **kwargs)
def build_sites_by_service_group_delete_request( # pylint: disable=name-too-long
servicegroup_name: str, site_name: 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", "2025-03-01-preview"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/providers/Microsoft.Management/serviceGroups/{servicegroupName}/providers/Microsoft.Edge/sites/{siteName}"
path_format_arguments = {
"servicegroupName": _SERIALIZER.url("servicegroup_name", servicegroup_name, "str"),
"siteName": _SERIALIZER.url("site_name", site_name, "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="DELETE", url=_url, params=_params, headers=_headers, **kwargs)
[docs]
class SitesOperations:
"""
.. warning::
**DO NOT** instantiate this class directly.
Instead, you should access the following operations through
:class:`~azure.mgmt.sitemanager.SiteManagerMgmtClient`'s
:attr:`sites` attribute.
"""
def __init__(self, *args, **kwargs) -> None:
input_args = list(args)
self._client: PipelineClient = input_args.pop(0) if input_args else kwargs.pop("client")
self._config: SiteManagerMgmtClientConfiguration = input_args.pop(0) if input_args else kwargs.pop("config")
self._serialize: Serializer = input_args.pop(0) if input_args else kwargs.pop("serializer")
self._deserialize: Deserializer = input_args.pop(0) if input_args else kwargs.pop("deserializer")
[docs]
@distributed_trace
def get(self, resource_group_name: str, site_name: str, **kwargs: Any) -> _models.Site:
"""Get a Site.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param site_name: Name of Site resource. Required.
:type site_name: str
:return: Site. The Site is compatible with MutableMapping
:rtype: ~azure.mgmt.sitemanager.models.Site
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = kwargs.pop("headers", {}) or {}
_params = kwargs.pop("params", {}) or {}
cls: ClsType[_models.Site] = kwargs.pop("cls", None)
_request = build_sites_get_request(
resource_group_name=resource_group_name,
site_name=site_name,
subscription_id=self._config.subscription_id,
api_version=self._config.api_version,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
_stream = kwargs.pop("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]:
if _stream:
try:
response.read() # Load the body in memory and close the socket
except (StreamConsumedError, StreamClosedError):
pass
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
if _stream:
deserialized = response.iter_bytes()
else:
deserialized = _deserialize(_models.Site, response.json())
if cls:
return cls(pipeline_response, deserialized, {}) # type: ignore
return deserialized # type: ignore
def _create_or_update_initial(
self, resource_group_name: str, site_name: str, resource: Union[_models.Site, JSON, IO[bytes]], **kwargs: Any
) -> Iterator[bytes]:
error_map: MutableMapping = {
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 = kwargs.pop("params", {}) or {}
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None)
content_type = content_type or "application/json"
_content = None
if isinstance(resource, (IOBase, bytes)):
_content = resource
else:
_content = json.dumps(resource, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore
_request = build_sites_create_or_update_request(
resource_group_name=resource_group_name,
site_name=site_name,
subscription_id=self._config.subscription_id,
content_type=content_type,
api_version=self._config.api_version,
content=_content,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
_stream = True
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, 201]:
try:
response.read() # Load the body in memory and close the socket
except (StreamConsumedError, StreamClosedError):
pass
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
response_headers = {}
if response.status_code == 201:
response_headers["Azure-AsyncOperation"] = self._deserialize(
"str", response.headers.get("Azure-AsyncOperation")
)
response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After"))
deserialized = response.iter_bytes()
if cls:
return cls(pipeline_response, deserialized, response_headers) # type: ignore
return deserialized # type: ignore
@overload
def begin_create_or_update(
self,
resource_group_name: str,
site_name: str,
resource: _models.Site,
*,
content_type: str = "application/json",
**kwargs: Any
) -> LROPoller[_models.Site]:
"""Create a Site.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param site_name: Name of Site resource. Required.
:type site_name: str
:param resource: Resource create parameters. Required.
:type resource: ~azure.mgmt.sitemanager.models.Site
: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 Site. The Site is compatible with MutableMapping
:rtype: ~azure.core.polling.LROPoller[~azure.mgmt.sitemanager.models.Site]
:raises ~azure.core.exceptions.HttpResponseError:
"""
@overload
def begin_create_or_update(
self,
resource_group_name: str,
site_name: str,
resource: JSON,
*,
content_type: str = "application/json",
**kwargs: Any
) -> LROPoller[_models.Site]:
"""Create a Site.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param site_name: Name of Site resource. Required.
:type site_name: str
:param resource: Resource create parameters. Required.
:type resource: JSON
: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 Site. The Site is compatible with MutableMapping
:rtype: ~azure.core.polling.LROPoller[~azure.mgmt.sitemanager.models.Site]
:raises ~azure.core.exceptions.HttpResponseError:
"""
@overload
def begin_create_or_update(
self,
resource_group_name: str,
site_name: str,
resource: IO[bytes],
*,
content_type: str = "application/json",
**kwargs: Any
) -> LROPoller[_models.Site]:
"""Create a Site.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param site_name: Name of Site resource. Required.
:type site_name: str
:param resource: Resource create parameters. Required.
:type resource: 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 Site. The Site is compatible with MutableMapping
:rtype: ~azure.core.polling.LROPoller[~azure.mgmt.sitemanager.models.Site]
:raises ~azure.core.exceptions.HttpResponseError:
"""
[docs]
@distributed_trace
def begin_create_or_update(
self, resource_group_name: str, site_name: str, resource: Union[_models.Site, JSON, IO[bytes]], **kwargs: Any
) -> LROPoller[_models.Site]:
"""Create a Site.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param site_name: Name of Site resource. Required.
:type site_name: str
:param resource: Resource create parameters. Is one of the following types: Site, JSON,
IO[bytes] Required.
:type resource: ~azure.mgmt.sitemanager.models.Site or JSON or IO[bytes]
:return: An instance of LROPoller that returns Site. The Site is compatible with MutableMapping
:rtype: ~azure.core.polling.LROPoller[~azure.mgmt.sitemanager.models.Site]
:raises ~azure.core.exceptions.HttpResponseError:
"""
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = kwargs.pop("params", {}) or {}
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
cls: ClsType[_models.Site] = 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_or_update_initial(
resource_group_name=resource_group_name,
site_name=site_name,
resource=resource,
content_type=content_type,
cls=lambda x, y, z: x,
headers=_headers,
params=_params,
**kwargs
)
raw_result.http_response.read() # type: ignore
kwargs.pop("error_map", None)
def get_long_running_output(pipeline_response):
response = pipeline_response.http_response
deserialized = _deserialize(_models.Site, response.json())
if cls:
return cls(pipeline_response, deserialized, {}) # type: ignore
return deserialized
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
if polling is True:
polling_method: PollingMethod = cast(
PollingMethod, ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs)
)
elif polling is False:
polling_method = cast(PollingMethod, NoPolling())
else:
polling_method = polling
if cont_token:
return LROPoller[_models.Site].from_continuation_token(
polling_method=polling_method,
continuation_token=cont_token,
client=self._client,
deserialization_callback=get_long_running_output,
)
return LROPoller[_models.Site](
self._client, raw_result, get_long_running_output, polling_method # type: ignore
)
@overload
def update(
self,
resource_group_name: str,
site_name: str,
properties: _models.SiteUpdate,
*,
content_type: str = "application/json",
**kwargs: Any
) -> _models.Site:
"""Update a Site.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param site_name: Name of Site resource. Required.
:type site_name: str
:param properties: The resource properties to be updated. Required.
:type properties: ~azure.mgmt.sitemanager.models.SiteUpdate
:keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
Default value is "application/json".
:paramtype content_type: str
:return: Site. The Site is compatible with MutableMapping
:rtype: ~azure.mgmt.sitemanager.models.Site
:raises ~azure.core.exceptions.HttpResponseError:
"""
@overload
def update(
self,
resource_group_name: str,
site_name: str,
properties: JSON,
*,
content_type: str = "application/json",
**kwargs: Any
) -> _models.Site:
"""Update a Site.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param site_name: Name of Site resource. Required.
:type site_name: str
:param properties: The resource properties to be updated. Required.
:type properties: JSON
:keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
Default value is "application/json".
:paramtype content_type: str
:return: Site. The Site is compatible with MutableMapping
:rtype: ~azure.mgmt.sitemanager.models.Site
:raises ~azure.core.exceptions.HttpResponseError:
"""
@overload
def update(
self,
resource_group_name: str,
site_name: str,
properties: IO[bytes],
*,
content_type: str = "application/json",
**kwargs: Any
) -> _models.Site:
"""Update a Site.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param site_name: Name of Site resource. Required.
:type site_name: str
:param properties: The resource properties to be updated. Required.
:type properties: 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: Site. The Site is compatible with MutableMapping
:rtype: ~azure.mgmt.sitemanager.models.Site
:raises ~azure.core.exceptions.HttpResponseError:
"""
[docs]
@distributed_trace
def update(
self,
resource_group_name: str,
site_name: str,
properties: Union[_models.SiteUpdate, JSON, IO[bytes]],
**kwargs: Any
) -> _models.Site:
"""Update a Site.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param site_name: Name of Site resource. Required.
:type site_name: str
:param properties: The resource properties to be updated. Is one of the following types:
SiteUpdate, JSON, IO[bytes] Required.
:type properties: ~azure.mgmt.sitemanager.models.SiteUpdate or JSON or IO[bytes]
:return: Site. The Site is compatible with MutableMapping
:rtype: ~azure.mgmt.sitemanager.models.Site
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
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 = kwargs.pop("params", {}) or {}
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
cls: ClsType[_models.Site] = kwargs.pop("cls", None)
content_type = content_type or "application/json"
_content = None
if isinstance(properties, (IOBase, bytes)):
_content = properties
else:
_content = json.dumps(properties, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore
_request = build_sites_update_request(
resource_group_name=resource_group_name,
site_name=site_name,
subscription_id=self._config.subscription_id,
content_type=content_type,
api_version=self._config.api_version,
content=_content,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
_stream = kwargs.pop("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]:
if _stream:
try:
response.read() # Load the body in memory and close the socket
except (StreamConsumedError, StreamClosedError):
pass
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
if _stream:
deserialized = response.iter_bytes()
else:
deserialized = _deserialize(_models.Site, response.json())
if cls:
return cls(pipeline_response, deserialized, {}) # type: ignore
return deserialized # type: ignore
[docs]
@distributed_trace
def delete( # pylint: disable=inconsistent-return-statements
self, resource_group_name: str, site_name: str, **kwargs: Any
) -> None:
"""Delete a Site.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param site_name: Name of Site resource. Required.
:type site_name: str
:return: None
:rtype: None
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = kwargs.pop("headers", {}) or {}
_params = kwargs.pop("params", {}) or {}
cls: ClsType[None] = kwargs.pop("cls", None)
_request = build_sites_delete_request(
resource_group_name=resource_group_name,
site_name=site_name,
subscription_id=self._config.subscription_id,
api_version=self._config.api_version,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
_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, 204]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
if cls:
return cls(pipeline_response, None, {}) # type: ignore
[docs]
@distributed_trace
def list_by_resource_group(self, resource_group_name: str, **kwargs: Any) -> Iterable["_models.Site"]:
"""List Site resources by resource group.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:return: An iterator like instance of Site
:rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.sitemanager.models.Site]
:raises ~azure.core.exceptions.HttpResponseError:
"""
_headers = kwargs.pop("headers", {}) or {}
_params = kwargs.pop("params", {}) or {}
cls: ClsType[List[_models.Site]] = kwargs.pop("cls", None)
error_map: MutableMapping = {
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_sites_list_by_resource_group_request(
resource_group_name=resource_group_name,
subscription_id=self._config.subscription_id,
api_version=self._config.api_version,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url(
"self._config.base_url", self._config.base_url, "str", skip_quote=True
),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
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
)
path_format_arguments = {
"endpoint": self._serialize.url(
"self._config.base_url", self._config.base_url, "str", skip_quote=True
),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
return _request
def extract_data(pipeline_response):
deserialized = pipeline_response.http_response.json()
list_of_elem = _deserialize(List[_models.Site], deserialized.get("value", []))
if cls:
list_of_elem = cls(list_of_elem) # type: ignore
return deserialized.get("nextLink") 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 = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
return pipeline_response
return ItemPaged(get_next, extract_data)
[docs]
class SitesBySubscriptionOperations:
"""
.. warning::
**DO NOT** instantiate this class directly.
Instead, you should access the following operations through
:class:`~azure.mgmt.sitemanager.SiteManagerMgmtClient`'s
:attr:`sites_by_subscription` attribute.
"""
def __init__(self, *args, **kwargs) -> None:
input_args = list(args)
self._client: PipelineClient = input_args.pop(0) if input_args else kwargs.pop("client")
self._config: SiteManagerMgmtClientConfiguration = input_args.pop(0) if input_args else kwargs.pop("config")
self._serialize: Serializer = input_args.pop(0) if input_args else kwargs.pop("serializer")
self._deserialize: Deserializer = input_args.pop(0) if input_args else kwargs.pop("deserializer")
[docs]
@distributed_trace
def list(self, **kwargs: Any) -> Iterable["_models.Site"]:
"""List Site resources by subscription ID.
:return: An iterator like instance of Site
:rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.sitemanager.models.Site]
:raises ~azure.core.exceptions.HttpResponseError:
"""
_headers = kwargs.pop("headers", {}) or {}
_params = kwargs.pop("params", {}) or {}
cls: ClsType[List[_models.Site]] = kwargs.pop("cls", None)
error_map: MutableMapping = {
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_sites_by_subscription_list_request(
subscription_id=self._config.subscription_id,
api_version=self._config.api_version,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url(
"self._config.base_url", self._config.base_url, "str", skip_quote=True
),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
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
)
path_format_arguments = {
"endpoint": self._serialize.url(
"self._config.base_url", self._config.base_url, "str", skip_quote=True
),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
return _request
def extract_data(pipeline_response):
deserialized = pipeline_response.http_response.json()
list_of_elem = _deserialize(List[_models.Site], deserialized.get("value", []))
if cls:
list_of_elem = cls(list_of_elem) # type: ignore
return deserialized.get("nextLink") 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 = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
return pipeline_response
return ItemPaged(get_next, extract_data)
[docs]
@distributed_trace
def get(self, site_name: str, **kwargs: Any) -> _models.Site:
"""Get a Site.
:param site_name: Name of Site resource. Required.
:type site_name: str
:return: Site. The Site is compatible with MutableMapping
:rtype: ~azure.mgmt.sitemanager.models.Site
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = kwargs.pop("headers", {}) or {}
_params = kwargs.pop("params", {}) or {}
cls: ClsType[_models.Site] = kwargs.pop("cls", None)
_request = build_sites_by_subscription_get_request(
site_name=site_name,
subscription_id=self._config.subscription_id,
api_version=self._config.api_version,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
_stream = kwargs.pop("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]:
if _stream:
try:
response.read() # Load the body in memory and close the socket
except (StreamConsumedError, StreamClosedError):
pass
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
if _stream:
deserialized = response.iter_bytes()
else:
deserialized = _deserialize(_models.Site, response.json())
if cls:
return cls(pipeline_response, deserialized, {}) # type: ignore
return deserialized # type: ignore
def _create_or_update_initial(
self, site_name: str, resource: Union[_models.Site, JSON, IO[bytes]], **kwargs: Any
) -> Iterator[bytes]:
error_map: MutableMapping = {
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 = kwargs.pop("params", {}) or {}
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None)
content_type = content_type or "application/json"
_content = None
if isinstance(resource, (IOBase, bytes)):
_content = resource
else:
_content = json.dumps(resource, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore
_request = build_sites_by_subscription_create_or_update_request(
site_name=site_name,
subscription_id=self._config.subscription_id,
content_type=content_type,
api_version=self._config.api_version,
content=_content,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
_stream = True
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, 201]:
try:
response.read() # Load the body in memory and close the socket
except (StreamConsumedError, StreamClosedError):
pass
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
response_headers = {}
if response.status_code == 201:
response_headers["Azure-AsyncOperation"] = self._deserialize(
"str", response.headers.get("Azure-AsyncOperation")
)
response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After"))
deserialized = response.iter_bytes()
if cls:
return cls(pipeline_response, deserialized, response_headers) # type: ignore
return deserialized # type: ignore
@overload
def begin_create_or_update(
self, site_name: str, resource: _models.Site, *, content_type: str = "application/json", **kwargs: Any
) -> LROPoller[_models.Site]:
"""Create a Site.
:param site_name: Name of Site resource. Required.
:type site_name: str
:param resource: Resource create parameters. Required.
:type resource: ~azure.mgmt.sitemanager.models.Site
: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 Site. The Site is compatible with MutableMapping
:rtype: ~azure.core.polling.LROPoller[~azure.mgmt.sitemanager.models.Site]
:raises ~azure.core.exceptions.HttpResponseError:
"""
@overload
def begin_create_or_update(
self, site_name: str, resource: JSON, *, content_type: str = "application/json", **kwargs: Any
) -> LROPoller[_models.Site]:
"""Create a Site.
:param site_name: Name of Site resource. Required.
:type site_name: str
:param resource: Resource create parameters. Required.
:type resource: JSON
: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 Site. The Site is compatible with MutableMapping
:rtype: ~azure.core.polling.LROPoller[~azure.mgmt.sitemanager.models.Site]
:raises ~azure.core.exceptions.HttpResponseError:
"""
@overload
def begin_create_or_update(
self, site_name: str, resource: IO[bytes], *, content_type: str = "application/json", **kwargs: Any
) -> LROPoller[_models.Site]:
"""Create a Site.
:param site_name: Name of Site resource. Required.
:type site_name: str
:param resource: Resource create parameters. Required.
:type resource: 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 Site. The Site is compatible with MutableMapping
:rtype: ~azure.core.polling.LROPoller[~azure.mgmt.sitemanager.models.Site]
:raises ~azure.core.exceptions.HttpResponseError:
"""
[docs]
@distributed_trace
def begin_create_or_update(
self, site_name: str, resource: Union[_models.Site, JSON, IO[bytes]], **kwargs: Any
) -> LROPoller[_models.Site]:
"""Create a Site.
:param site_name: Name of Site resource. Required.
:type site_name: str
:param resource: Resource create parameters. Is one of the following types: Site, JSON,
IO[bytes] Required.
:type resource: ~azure.mgmt.sitemanager.models.Site or JSON or IO[bytes]
:return: An instance of LROPoller that returns Site. The Site is compatible with MutableMapping
:rtype: ~azure.core.polling.LROPoller[~azure.mgmt.sitemanager.models.Site]
:raises ~azure.core.exceptions.HttpResponseError:
"""
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = kwargs.pop("params", {}) or {}
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
cls: ClsType[_models.Site] = 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_or_update_initial(
site_name=site_name,
resource=resource,
content_type=content_type,
cls=lambda x, y, z: x,
headers=_headers,
params=_params,
**kwargs
)
raw_result.http_response.read() # type: ignore
kwargs.pop("error_map", None)
def get_long_running_output(pipeline_response):
response = pipeline_response.http_response
deserialized = _deserialize(_models.Site, response.json())
if cls:
return cls(pipeline_response, deserialized, {}) # type: ignore
return deserialized
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
if polling is True:
polling_method: PollingMethod = cast(
PollingMethod, ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs)
)
elif polling is False:
polling_method = cast(PollingMethod, NoPolling())
else:
polling_method = polling
if cont_token:
return LROPoller[_models.Site].from_continuation_token(
polling_method=polling_method,
continuation_token=cont_token,
client=self._client,
deserialization_callback=get_long_running_output,
)
return LROPoller[_models.Site](
self._client, raw_result, get_long_running_output, polling_method # type: ignore
)
@overload
def update(
self, site_name: str, properties: _models.SiteUpdate, *, content_type: str = "application/json", **kwargs: Any
) -> _models.Site:
"""Update a Site.
:param site_name: Name of Site resource. Required.
:type site_name: str
:param properties: The resource properties to be updated. Required.
:type properties: ~azure.mgmt.sitemanager.models.SiteUpdate
:keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
Default value is "application/json".
:paramtype content_type: str
:return: Site. The Site is compatible with MutableMapping
:rtype: ~azure.mgmt.sitemanager.models.Site
:raises ~azure.core.exceptions.HttpResponseError:
"""
@overload
def update(
self, site_name: str, properties: JSON, *, content_type: str = "application/json", **kwargs: Any
) -> _models.Site:
"""Update a Site.
:param site_name: Name of Site resource. Required.
:type site_name: str
:param properties: The resource properties to be updated. Required.
:type properties: JSON
:keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
Default value is "application/json".
:paramtype content_type: str
:return: Site. The Site is compatible with MutableMapping
:rtype: ~azure.mgmt.sitemanager.models.Site
:raises ~azure.core.exceptions.HttpResponseError:
"""
@overload
def update(
self, site_name: str, properties: IO[bytes], *, content_type: str = "application/json", **kwargs: Any
) -> _models.Site:
"""Update a Site.
:param site_name: Name of Site resource. Required.
:type site_name: str
:param properties: The resource properties to be updated. Required.
:type properties: 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: Site. The Site is compatible with MutableMapping
:rtype: ~azure.mgmt.sitemanager.models.Site
:raises ~azure.core.exceptions.HttpResponseError:
"""
[docs]
@distributed_trace
def update(
self, site_name: str, properties: Union[_models.SiteUpdate, JSON, IO[bytes]], **kwargs: Any
) -> _models.Site:
"""Update a Site.
:param site_name: Name of Site resource. Required.
:type site_name: str
:param properties: The resource properties to be updated. Is one of the following types:
SiteUpdate, JSON, IO[bytes] Required.
:type properties: ~azure.mgmt.sitemanager.models.SiteUpdate or JSON or IO[bytes]
:return: Site. The Site is compatible with MutableMapping
:rtype: ~azure.mgmt.sitemanager.models.Site
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
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 = kwargs.pop("params", {}) or {}
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
cls: ClsType[_models.Site] = kwargs.pop("cls", None)
content_type = content_type or "application/json"
_content = None
if isinstance(properties, (IOBase, bytes)):
_content = properties
else:
_content = json.dumps(properties, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore
_request = build_sites_by_subscription_update_request(
site_name=site_name,
subscription_id=self._config.subscription_id,
content_type=content_type,
api_version=self._config.api_version,
content=_content,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
_stream = kwargs.pop("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]:
if _stream:
try:
response.read() # Load the body in memory and close the socket
except (StreamConsumedError, StreamClosedError):
pass
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
if _stream:
deserialized = response.iter_bytes()
else:
deserialized = _deserialize(_models.Site, response.json())
if cls:
return cls(pipeline_response, deserialized, {}) # type: ignore
return deserialized # type: ignore
[docs]
@distributed_trace
def delete(self, site_name: str, **kwargs: Any) -> None: # pylint: disable=inconsistent-return-statements
"""Delete a Site.
:param site_name: Name of Site resource. Required.
:type site_name: str
:return: None
:rtype: None
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = kwargs.pop("headers", {}) or {}
_params = kwargs.pop("params", {}) or {}
cls: ClsType[None] = kwargs.pop("cls", None)
_request = build_sites_by_subscription_delete_request(
site_name=site_name,
subscription_id=self._config.subscription_id,
api_version=self._config.api_version,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
_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, 204]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
if cls:
return cls(pipeline_response, None, {}) # type: ignore
[docs]
class SitesByServiceGroupOperations:
"""
.. warning::
**DO NOT** instantiate this class directly.
Instead, you should access the following operations through
:class:`~azure.mgmt.sitemanager.SiteManagerMgmtClient`'s
:attr:`sites_by_service_group` attribute.
"""
def __init__(self, *args, **kwargs) -> None:
input_args = list(args)
self._client: PipelineClient = input_args.pop(0) if input_args else kwargs.pop("client")
self._config: SiteManagerMgmtClientConfiguration = input_args.pop(0) if input_args else kwargs.pop("config")
self._serialize: Serializer = input_args.pop(0) if input_args else kwargs.pop("serializer")
self._deserialize: Deserializer = input_args.pop(0) if input_args else kwargs.pop("deserializer")
[docs]
@distributed_trace
@api_version_validation(
method_added_on="2025-03-01-preview",
params_added_on={"2025-03-01-preview": ["api_version", "servicegroup_name", "accept"]},
)
def list_by_service_group(self, servicegroup_name: str, **kwargs: Any) -> Iterable["_models.Site"]:
"""list Site at SG scope.
:param servicegroup_name: The name of the service group. Required.
:type servicegroup_name: str
:return: An iterator like instance of Site
:rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.sitemanager.models.Site]
:raises ~azure.core.exceptions.HttpResponseError:
"""
_headers = kwargs.pop("headers", {}) or {}
_params = kwargs.pop("params", {}) or {}
cls: ClsType[List[_models.Site]] = kwargs.pop("cls", None)
error_map: MutableMapping = {
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_sites_by_service_group_list_by_service_group_request(
servicegroup_name=servicegroup_name,
api_version=self._config.api_version,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url(
"self._config.base_url", self._config.base_url, "str", skip_quote=True
),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
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
)
path_format_arguments = {
"endpoint": self._serialize.url(
"self._config.base_url", self._config.base_url, "str", skip_quote=True
),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
return _request
def extract_data(pipeline_response):
deserialized = pipeline_response.http_response.json()
list_of_elem = _deserialize(List[_models.Site], deserialized.get("value", []))
if cls:
list_of_elem = cls(list_of_elem) # type: ignore
return deserialized.get("nextLink") 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 = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
return pipeline_response
return ItemPaged(get_next, extract_data)
[docs]
@distributed_trace
@api_version_validation(
method_added_on="2025-03-01-preview",
params_added_on={"2025-03-01-preview": ["api_version", "servicegroup_name", "site_name", "accept"]},
)
def get(self, servicegroup_name: str, site_name: str, **kwargs: Any) -> _models.Site:
"""Get Site at SG scope.
:param servicegroup_name: The name of the service group. Required.
:type servicegroup_name: str
:param site_name: The name of the site. Required.
:type site_name: str
:return: Site. The Site is compatible with MutableMapping
:rtype: ~azure.mgmt.sitemanager.models.Site
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = kwargs.pop("headers", {}) or {}
_params = kwargs.pop("params", {}) or {}
cls: ClsType[_models.Site] = kwargs.pop("cls", None)
_request = build_sites_by_service_group_get_request(
servicegroup_name=servicegroup_name,
site_name=site_name,
api_version=self._config.api_version,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
_stream = kwargs.pop("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]:
if _stream:
try:
response.read() # Load the body in memory and close the socket
except (StreamConsumedError, StreamClosedError):
pass
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
if _stream:
deserialized = response.iter_bytes()
else:
deserialized = _deserialize(_models.Site, response.json())
if cls:
return cls(pipeline_response, deserialized, {}) # type: ignore
return deserialized # type: ignore
@api_version_validation(
method_added_on="2025-03-01-preview",
params_added_on={
"2025-03-01-preview": ["api_version", "servicegroup_name", "site_name", "content_type", "accept"]
},
)
def _create_or_update_initial(
self, servicegroup_name: str, site_name: str, resource: Union[_models.Site, JSON, IO[bytes]], **kwargs: Any
) -> Iterator[bytes]:
error_map: MutableMapping = {
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 = kwargs.pop("params", {}) or {}
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None)
content_type = content_type or "application/json"
_content = None
if isinstance(resource, (IOBase, bytes)):
_content = resource
else:
_content = json.dumps(resource, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore
_request = build_sites_by_service_group_create_or_update_request(
servicegroup_name=servicegroup_name,
site_name=site_name,
content_type=content_type,
api_version=self._config.api_version,
content=_content,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
_stream = True
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, 201]:
try:
response.read() # Load the body in memory and close the socket
except (StreamConsumedError, StreamClosedError):
pass
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
response_headers = {}
if response.status_code == 201:
response_headers["Azure-AsyncOperation"] = self._deserialize(
"str", response.headers.get("Azure-AsyncOperation")
)
response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After"))
deserialized = response.iter_bytes()
if cls:
return cls(pipeline_response, deserialized, response_headers) # type: ignore
return deserialized # type: ignore
@overload
def begin_create_or_update(
self,
servicegroup_name: str,
site_name: str,
resource: _models.Site,
*,
content_type: str = "application/json",
**kwargs: Any
) -> LROPoller[_models.Site]:
"""create or update Site at SG scope.
:param servicegroup_name: The name of the service group. Required.
:type servicegroup_name: str
:param site_name: The name of the site. Required.
:type site_name: str
:param resource: The properties of the site. Required.
:type resource: ~azure.mgmt.sitemanager.models.Site
: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 Site. The Site is compatible with MutableMapping
:rtype: ~azure.core.polling.LROPoller[~azure.mgmt.sitemanager.models.Site]
:raises ~azure.core.exceptions.HttpResponseError:
"""
@overload
def begin_create_or_update(
self,
servicegroup_name: str,
site_name: str,
resource: JSON,
*,
content_type: str = "application/json",
**kwargs: Any
) -> LROPoller[_models.Site]:
"""create or update Site at SG scope.
:param servicegroup_name: The name of the service group. Required.
:type servicegroup_name: str
:param site_name: The name of the site. Required.
:type site_name: str
:param resource: The properties of the site. Required.
:type resource: JSON
: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 Site. The Site is compatible with MutableMapping
:rtype: ~azure.core.polling.LROPoller[~azure.mgmt.sitemanager.models.Site]
:raises ~azure.core.exceptions.HttpResponseError:
"""
@overload
def begin_create_or_update(
self,
servicegroup_name: str,
site_name: str,
resource: IO[bytes],
*,
content_type: str = "application/json",
**kwargs: Any
) -> LROPoller[_models.Site]:
"""create or update Site at SG scope.
:param servicegroup_name: The name of the service group. Required.
:type servicegroup_name: str
:param site_name: The name of the site. Required.
:type site_name: str
:param resource: The properties of the site. Required.
:type resource: 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 Site. The Site is compatible with MutableMapping
:rtype: ~azure.core.polling.LROPoller[~azure.mgmt.sitemanager.models.Site]
:raises ~azure.core.exceptions.HttpResponseError:
"""
[docs]
@distributed_trace
@api_version_validation(
method_added_on="2025-03-01-preview",
params_added_on={
"2025-03-01-preview": ["api_version", "servicegroup_name", "site_name", "content_type", "accept"]
},
)
def begin_create_or_update(
self, servicegroup_name: str, site_name: str, resource: Union[_models.Site, JSON, IO[bytes]], **kwargs: Any
) -> LROPoller[_models.Site]:
"""create or update Site at SG scope.
:param servicegroup_name: The name of the service group. Required.
:type servicegroup_name: str
:param site_name: The name of the site. Required.
:type site_name: str
:param resource: The properties of the site. Is one of the following types: Site, JSON,
IO[bytes] Required.
:type resource: ~azure.mgmt.sitemanager.models.Site or JSON or IO[bytes]
:return: An instance of LROPoller that returns Site. The Site is compatible with MutableMapping
:rtype: ~azure.core.polling.LROPoller[~azure.mgmt.sitemanager.models.Site]
:raises ~azure.core.exceptions.HttpResponseError:
"""
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = kwargs.pop("params", {}) or {}
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
cls: ClsType[_models.Site] = 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_or_update_initial(
servicegroup_name=servicegroup_name,
site_name=site_name,
resource=resource,
content_type=content_type,
cls=lambda x, y, z: x,
headers=_headers,
params=_params,
**kwargs
)
raw_result.http_response.read() # type: ignore
kwargs.pop("error_map", None)
def get_long_running_output(pipeline_response):
response = pipeline_response.http_response
deserialized = _deserialize(_models.Site, response.json())
if cls:
return cls(pipeline_response, deserialized, {}) # type: ignore
return deserialized
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
if polling is True:
polling_method: PollingMethod = cast(
PollingMethod, ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs)
)
elif polling is False:
polling_method = cast(PollingMethod, NoPolling())
else:
polling_method = polling
if cont_token:
return LROPoller[_models.Site].from_continuation_token(
polling_method=polling_method,
continuation_token=cont_token,
client=self._client,
deserialization_callback=get_long_running_output,
)
return LROPoller[_models.Site](
self._client, raw_result, get_long_running_output, polling_method # type: ignore
)
@overload
def update(
self,
servicegroup_name: str,
site_name: str,
properties: _models.SiteUpdate,
*,
content_type: str = "application/json",
**kwargs: Any
) -> _models.Site:
"""update Site at SG scope.
:param servicegroup_name: The name of the service group. Required.
:type servicegroup_name: str
:param site_name: The name of the site. Required.
:type site_name: str
:param properties: The properties of the site. Required.
:type properties: ~azure.mgmt.sitemanager.models.SiteUpdate
:keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
Default value is "application/json".
:paramtype content_type: str
:return: Site. The Site is compatible with MutableMapping
:rtype: ~azure.mgmt.sitemanager.models.Site
:raises ~azure.core.exceptions.HttpResponseError:
"""
@overload
def update(
self,
servicegroup_name: str,
site_name: str,
properties: JSON,
*,
content_type: str = "application/json",
**kwargs: Any
) -> _models.Site:
"""update Site at SG scope.
:param servicegroup_name: The name of the service group. Required.
:type servicegroup_name: str
:param site_name: The name of the site. Required.
:type site_name: str
:param properties: The properties of the site. Required.
:type properties: JSON
:keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
Default value is "application/json".
:paramtype content_type: str
:return: Site. The Site is compatible with MutableMapping
:rtype: ~azure.mgmt.sitemanager.models.Site
:raises ~azure.core.exceptions.HttpResponseError:
"""
@overload
def update(
self,
servicegroup_name: str,
site_name: str,
properties: IO[bytes],
*,
content_type: str = "application/json",
**kwargs: Any
) -> _models.Site:
"""update Site at SG scope.
:param servicegroup_name: The name of the service group. Required.
:type servicegroup_name: str
:param site_name: The name of the site. Required.
:type site_name: str
:param properties: The properties of the site. Required.
:type properties: 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: Site. The Site is compatible with MutableMapping
:rtype: ~azure.mgmt.sitemanager.models.Site
:raises ~azure.core.exceptions.HttpResponseError:
"""
[docs]
@distributed_trace
@api_version_validation(
method_added_on="2025-03-01-preview",
params_added_on={
"2025-03-01-preview": ["api_version", "servicegroup_name", "site_name", "content_type", "accept"]
},
)
def update(
self,
servicegroup_name: str,
site_name: str,
properties: Union[_models.SiteUpdate, JSON, IO[bytes]],
**kwargs: Any
) -> _models.Site:
"""update Site at SG scope.
:param servicegroup_name: The name of the service group. Required.
:type servicegroup_name: str
:param site_name: The name of the site. Required.
:type site_name: str
:param properties: The properties of the site. Is one of the following types: SiteUpdate, JSON,
IO[bytes] Required.
:type properties: ~azure.mgmt.sitemanager.models.SiteUpdate or JSON or IO[bytes]
:return: Site. The Site is compatible with MutableMapping
:rtype: ~azure.mgmt.sitemanager.models.Site
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
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 = kwargs.pop("params", {}) or {}
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
cls: ClsType[_models.Site] = kwargs.pop("cls", None)
content_type = content_type or "application/json"
_content = None
if isinstance(properties, (IOBase, bytes)):
_content = properties
else:
_content = json.dumps(properties, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore
_request = build_sites_by_service_group_update_request(
servicegroup_name=servicegroup_name,
site_name=site_name,
content_type=content_type,
api_version=self._config.api_version,
content=_content,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
_stream = kwargs.pop("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]:
if _stream:
try:
response.read() # Load the body in memory and close the socket
except (StreamConsumedError, StreamClosedError):
pass
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
if _stream:
deserialized = response.iter_bytes()
else:
deserialized = _deserialize(_models.Site, response.json())
if cls:
return cls(pipeline_response, deserialized, {}) # type: ignore
return deserialized # type: ignore
[docs]
@distributed_trace
@api_version_validation(
method_added_on="2025-03-01-preview",
params_added_on={"2025-03-01-preview": ["api_version", "servicegroup_name", "site_name", "accept"]},
)
def delete( # pylint: disable=inconsistent-return-statements
self, servicegroup_name: str, site_name: str, **kwargs: Any
) -> None:
"""delete Site at SG scope.
:param servicegroup_name: The name of the service group. Required.
:type servicegroup_name: str
:param site_name: The name of the site. Required.
:type site_name: str
:return: None
:rtype: None
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = kwargs.pop("headers", {}) or {}
_params = kwargs.pop("params", {}) or {}
cls: ClsType[None] = kwargs.pop("cls", None)
_request = build_sites_by_service_group_delete_request(
servicegroup_name=servicegroup_name,
site_name=site_name,
api_version=self._config.api_version,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
_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, 204]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
if cls:
return cls(pipeline_response, None, {}) # type: ignore