# 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 functools
from typing import TYPE_CHECKING
import warnings
from azure.core.exceptions import ClientAuthenticationError, HttpResponseError, ResourceExistsError, ResourceNotFoundError, map_error
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 msrest import Serializer
from .._vendor import _format_url_section
if TYPE_CHECKING:
# pylint: disable=unused-import,ungrouped-imports
from typing import Any, Callable, Dict, Generic, IO, List, Optional, TypeVar, Union
T = TypeVar('T')
ClsType = Optional[Callable[[PipelineResponse[HttpRequest, HttpResponse], T, Dict[str, Any]], Any]]
_SERIALIZER = Serializer()
# fmt: off
def build_generate_client_token_request(
hub, # type: str
**kwargs # type: Any
):
# type: (...) -> HttpRequest
user_id = kwargs.pop('user_id', None) # type: Optional[str]
role = kwargs.pop('role', None) # type: Optional[List[str]]
minutes_to_expire = kwargs.pop('minutes_to_expire', 60) # type: Optional[int]
api_version = "2021-10-01"
accept = "application/json, text/json"
# Construct URL
url = kwargs.pop("template_url", '/api/hubs/{hub}/:generateToken')
path_format_arguments = {
"hub": _SERIALIZER.url("hub", hub, 'str', pattern=r'^[A-Za-z][A-Za-z0-9_`,.[\]]{0,127}$'),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
if user_id is not None:
query_parameters['userId'] = _SERIALIZER.query("user_id", user_id, 'str')
if role is not None:
query_parameters['role'] = [_SERIALIZER.query("role", q, 'str') if q is not None else '' for q in role]
if minutes_to_expire is not None:
query_parameters['minutesToExpire'] = _SERIALIZER.query("minutes_to_expire", minutes_to_expire, 'int')
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="POST",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_close_all_connections_request(
hub, # type: str
**kwargs # type: Any
):
# type: (...) -> HttpRequest
excluded = kwargs.pop('excluded', None) # type: Optional[List[str]]
reason = kwargs.pop('reason', None) # type: Optional[str]
api_version = "2021-10-01"
accept = "application/json, text/json"
# Construct URL
url = kwargs.pop("template_url", '/api/hubs/{hub}/:closeConnections')
path_format_arguments = {
"hub": _SERIALIZER.url("hub", hub, 'str', pattern=r'^[A-Za-z][A-Za-z0-9_`,.[\]]{0,127}$'),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
if excluded is not None:
query_parameters['excluded'] = [_SERIALIZER.query("excluded", q, 'str') if q is not None else '' for q in excluded]
if reason is not None:
query_parameters['reason'] = _SERIALIZER.query("reason", reason, 'str')
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="POST",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_send_to_all_request(
hub, # type: str
**kwargs # type: Any
):
# type: (...) -> HttpRequest
content_type = kwargs.pop('content_type', None) # type: Optional[str]
excluded = kwargs.pop('excluded', None) # type: Optional[List[str]]
api_version = "2021-10-01"
accept = "application/json, text/json"
# Construct URL
url = kwargs.pop("template_url", '/api/hubs/{hub}/:send')
path_format_arguments = {
"hub": _SERIALIZER.url("hub", hub, 'str', pattern=r'^[A-Za-z][A-Za-z0-9_`,.[\]]{0,127}$'),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
if excluded is not None:
query_parameters['excluded'] = [_SERIALIZER.query("excluded", q, 'str') if q is not None else '' for q in excluded]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
if content_type is not None:
header_parameters['Content-Type'] = _SERIALIZER.header("content_type", content_type, 'str')
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="POST",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_connection_exists_request(
hub, # type: str
connection_id, # type: str
**kwargs # type: Any
):
# type: (...) -> HttpRequest
api_version = "2021-10-01"
accept = "application/json, text/json"
# Construct URL
url = kwargs.pop("template_url", '/api/hubs/{hub}/connections/{connectionId}')
path_format_arguments = {
"hub": _SERIALIZER.url("hub", hub, 'str', pattern=r'^[A-Za-z][A-Za-z0-9_`,.[\]]{0,127}$'),
"connectionId": _SERIALIZER.url("connection_id", connection_id, 'str', min_length=1),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="HEAD",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_close_connection_request(
hub, # type: str
connection_id, # type: str
**kwargs # type: Any
):
# type: (...) -> HttpRequest
reason = kwargs.pop('reason', None) # type: Optional[str]
api_version = "2021-10-01"
accept = "application/json, text/json"
# Construct URL
url = kwargs.pop("template_url", '/api/hubs/{hub}/connections/{connectionId}')
path_format_arguments = {
"hub": _SERIALIZER.url("hub", hub, 'str', pattern=r'^[A-Za-z][A-Za-z0-9_`,.[\]]{0,127}$'),
"connectionId": _SERIALIZER.url("connection_id", connection_id, 'str', min_length=1),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
if reason is not None:
query_parameters['reason'] = _SERIALIZER.query("reason", reason, 'str')
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="DELETE",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_send_to_connection_request(
hub, # type: str
connection_id, # type: str
**kwargs # type: Any
):
# type: (...) -> HttpRequest
content_type = kwargs.pop('content_type', None) # type: Optional[str]
api_version = "2021-10-01"
accept = "application/json, text/json"
# Construct URL
url = kwargs.pop("template_url", '/api/hubs/{hub}/connections/{connectionId}/:send')
path_format_arguments = {
"hub": _SERIALIZER.url("hub", hub, 'str', pattern=r'^[A-Za-z][A-Za-z0-9_`,.[\]]{0,127}$'),
"connectionId": _SERIALIZER.url("connection_id", connection_id, 'str', min_length=1),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
if content_type is not None:
header_parameters['Content-Type'] = _SERIALIZER.header("content_type", content_type, 'str')
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="POST",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_group_exists_request(
hub, # type: str
group, # type: str
**kwargs # type: Any
):
# type: (...) -> HttpRequest
api_version = "2021-10-01"
accept = "application/json, text/json"
# Construct URL
url = kwargs.pop("template_url", '/api/hubs/{hub}/groups/{group}')
path_format_arguments = {
"hub": _SERIALIZER.url("hub", hub, 'str', pattern=r'^[A-Za-z][A-Za-z0-9_`,.[\]]{0,127}$'),
"group": _SERIALIZER.url("group", group, 'str', max_length=1024, min_length=1),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="HEAD",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_close_group_connections_request(
hub, # type: str
group, # type: str
**kwargs # type: Any
):
# type: (...) -> HttpRequest
excluded = kwargs.pop('excluded', None) # type: Optional[List[str]]
reason = kwargs.pop('reason', None) # type: Optional[str]
api_version = "2021-10-01"
accept = "application/json, text/json"
# Construct URL
url = kwargs.pop("template_url", '/api/hubs/{hub}/groups/{group}/:closeConnections')
path_format_arguments = {
"hub": _SERIALIZER.url("hub", hub, 'str', pattern=r'^[A-Za-z][A-Za-z0-9_`,.[\]]{0,127}$'),
"group": _SERIALIZER.url("group", group, 'str', max_length=1024, min_length=1),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
if excluded is not None:
query_parameters['excluded'] = [_SERIALIZER.query("excluded", q, 'str') if q is not None else '' for q in excluded]
if reason is not None:
query_parameters['reason'] = _SERIALIZER.query("reason", reason, 'str')
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="POST",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_send_to_group_request(
hub, # type: str
group, # type: str
**kwargs # type: Any
):
# type: (...) -> HttpRequest
content_type = kwargs.pop('content_type', None) # type: Optional[str]
excluded = kwargs.pop('excluded', None) # type: Optional[List[str]]
api_version = "2021-10-01"
accept = "application/json, text/json"
# Construct URL
url = kwargs.pop("template_url", '/api/hubs/{hub}/groups/{group}/:send')
path_format_arguments = {
"hub": _SERIALIZER.url("hub", hub, 'str', pattern=r'^[A-Za-z][A-Za-z0-9_`,.[\]]{0,127}$'),
"group": _SERIALIZER.url("group", group, 'str', max_length=1024, min_length=1),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
if excluded is not None:
query_parameters['excluded'] = [_SERIALIZER.query("excluded", q, 'str') if q is not None else '' for q in excluded]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
if content_type is not None:
header_parameters['Content-Type'] = _SERIALIZER.header("content_type", content_type, 'str')
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="POST",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_add_connection_to_group_request(
hub, # type: str
group, # type: str
connection_id, # type: str
**kwargs # type: Any
):
# type: (...) -> HttpRequest
api_version = "2021-10-01"
accept = "application/json, text/json"
# Construct URL
url = kwargs.pop("template_url", '/api/hubs/{hub}/groups/{group}/connections/{connectionId}')
path_format_arguments = {
"hub": _SERIALIZER.url("hub", hub, 'str', pattern=r'^[A-Za-z][A-Za-z0-9_`,.[\]]{0,127}$'),
"group": _SERIALIZER.url("group", group, 'str', max_length=1024, min_length=1),
"connectionId": _SERIALIZER.url("connection_id", connection_id, 'str', min_length=1),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="PUT",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_remove_connection_from_group_request(
hub, # type: str
group, # type: str
connection_id, # type: str
**kwargs # type: Any
):
# type: (...) -> HttpRequest
api_version = "2021-10-01"
accept = "application/json, text/json"
# Construct URL
url = kwargs.pop("template_url", '/api/hubs/{hub}/groups/{group}/connections/{connectionId}')
path_format_arguments = {
"hub": _SERIALIZER.url("hub", hub, 'str', pattern=r'^[A-Za-z][A-Za-z0-9_`,.[\]]{0,127}$'),
"group": _SERIALIZER.url("group", group, 'str', max_length=1024, min_length=1),
"connectionId": _SERIALIZER.url("connection_id", connection_id, 'str', min_length=1),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="DELETE",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_user_exists_request(
hub, # type: str
user_id, # type: str
**kwargs # type: Any
):
# type: (...) -> HttpRequest
api_version = "2021-10-01"
accept = "application/json, text/json"
# Construct URL
url = kwargs.pop("template_url", '/api/hubs/{hub}/users/{userId}')
path_format_arguments = {
"hub": _SERIALIZER.url("hub", hub, 'str', pattern=r'^[A-Za-z][A-Za-z0-9_`,.[\]]{0,127}$'),
"userId": _SERIALIZER.url("user_id", user_id, 'str', min_length=1),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="HEAD",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_close_user_connections_request(
hub, # type: str
user_id, # type: str
**kwargs # type: Any
):
# type: (...) -> HttpRequest
excluded = kwargs.pop('excluded', None) # type: Optional[List[str]]
reason = kwargs.pop('reason', None) # type: Optional[str]
api_version = "2021-10-01"
accept = "application/json, text/json"
# Construct URL
url = kwargs.pop("template_url", '/api/hubs/{hub}/users/{userId}/:closeConnections')
path_format_arguments = {
"hub": _SERIALIZER.url("hub", hub, 'str', pattern=r'^[A-Za-z][A-Za-z0-9_`,.[\]]{0,127}$'),
"userId": _SERIALIZER.url("user_id", user_id, 'str', min_length=1),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
if excluded is not None:
query_parameters['excluded'] = [_SERIALIZER.query("excluded", q, 'str') if q is not None else '' for q in excluded]
if reason is not None:
query_parameters['reason'] = _SERIALIZER.query("reason", reason, 'str')
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="POST",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_send_to_user_request(
hub, # type: str
user_id, # type: str
**kwargs # type: Any
):
# type: (...) -> HttpRequest
content_type = kwargs.pop('content_type', None) # type: Optional[str]
api_version = "2021-10-01"
accept = "application/json, text/json"
# Construct URL
url = kwargs.pop("template_url", '/api/hubs/{hub}/users/{userId}/:send')
path_format_arguments = {
"hub": _SERIALIZER.url("hub", hub, 'str', pattern=r'^[A-Za-z][A-Za-z0-9_`,.[\]]{0,127}$'),
"userId": _SERIALIZER.url("user_id", user_id, 'str', min_length=1),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
if content_type is not None:
header_parameters['Content-Type'] = _SERIALIZER.header("content_type", content_type, 'str')
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="POST",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_add_user_to_group_request(
hub, # type: str
group, # type: str
user_id, # type: str
**kwargs # type: Any
):
# type: (...) -> HttpRequest
api_version = "2021-10-01"
accept = "application/json, text/json"
# Construct URL
url = kwargs.pop("template_url", '/api/hubs/{hub}/users/{userId}/groups/{group}')
path_format_arguments = {
"hub": _SERIALIZER.url("hub", hub, 'str', pattern=r'^[A-Za-z][A-Za-z0-9_`,.[\]]{0,127}$'),
"group": _SERIALIZER.url("group", group, 'str', max_length=1024, min_length=1),
"userId": _SERIALIZER.url("user_id", user_id, 'str', min_length=1),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="PUT",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_remove_user_from_group_request(
hub, # type: str
group, # type: str
user_id, # type: str
**kwargs # type: Any
):
# type: (...) -> HttpRequest
api_version = "2021-10-01"
accept = "application/json, text/json"
# Construct URL
url = kwargs.pop("template_url", '/api/hubs/{hub}/users/{userId}/groups/{group}')
path_format_arguments = {
"hub": _SERIALIZER.url("hub", hub, 'str', pattern=r'^[A-Za-z][A-Za-z0-9_`,.[\]]{0,127}$'),
"group": _SERIALIZER.url("group", group, 'str', max_length=1024, min_length=1),
"userId": _SERIALIZER.url("user_id", user_id, 'str', min_length=1),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="DELETE",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_remove_user_from_all_groups_request(
hub, # type: str
user_id, # type: str
**kwargs # type: Any
):
# type: (...) -> HttpRequest
api_version = "2021-10-01"
accept = "application/json, text/json"
# Construct URL
url = kwargs.pop("template_url", '/api/hubs/{hub}/users/{userId}/groups')
path_format_arguments = {
"hub": _SERIALIZER.url("hub", hub, 'str', pattern=r'^[A-Za-z][A-Za-z0-9_`,.[\]]{0,127}$'),
"userId": _SERIALIZER.url("user_id", user_id, 'str', min_length=1),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="DELETE",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_grant_permission_request(
hub, # type: str
permission, # type: str
connection_id, # type: str
**kwargs # type: Any
):
# type: (...) -> HttpRequest
target_name = kwargs.pop('target_name', None) # type: Optional[str]
api_version = "2021-10-01"
accept = "application/json, text/json"
# Construct URL
url = kwargs.pop("template_url", '/api/hubs/{hub}/permissions/{permission}/connections/{connectionId}')
path_format_arguments = {
"hub": _SERIALIZER.url("hub", hub, 'str', pattern=r'^[A-Za-z][A-Za-z0-9_`,.[\]]{0,127}$'),
"permission": _SERIALIZER.url("permission", permission, 'str'),
"connectionId": _SERIALIZER.url("connection_id", connection_id, 'str', min_length=1),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
if target_name is not None:
query_parameters['targetName'] = _SERIALIZER.query("target_name", target_name, 'str')
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="PUT",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_revoke_permission_request(
hub, # type: str
permission, # type: str
connection_id, # type: str
**kwargs # type: Any
):
# type: (...) -> HttpRequest
target_name = kwargs.pop('target_name', None) # type: Optional[str]
api_version = "2021-10-01"
accept = "application/json, text/json"
# Construct URL
url = kwargs.pop("template_url", '/api/hubs/{hub}/permissions/{permission}/connections/{connectionId}')
path_format_arguments = {
"hub": _SERIALIZER.url("hub", hub, 'str', pattern=r'^[A-Za-z][A-Za-z0-9_`,.[\]]{0,127}$'),
"permission": _SERIALIZER.url("permission", permission, 'str'),
"connectionId": _SERIALIZER.url("connection_id", connection_id, 'str', min_length=1),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
if target_name is not None:
query_parameters['targetName'] = _SERIALIZER.query("target_name", target_name, 'str')
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="DELETE",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_check_permission_request(
hub, # type: str
permission, # type: str
connection_id, # type: str
**kwargs # type: Any
):
# type: (...) -> HttpRequest
target_name = kwargs.pop('target_name', None) # type: Optional[str]
api_version = "2021-10-01"
accept = "application/json, text/json"
# Construct URL
url = kwargs.pop("template_url", '/api/hubs/{hub}/permissions/{permission}/connections/{connectionId}')
path_format_arguments = {
"hub": _SERIALIZER.url("hub", hub, 'str', pattern=r'^[A-Za-z][A-Za-z0-9_`,.[\]]{0,127}$'),
"permission": _SERIALIZER.url("permission", permission, 'str'),
"connectionId": _SERIALIZER.url("connection_id", connection_id, 'str', min_length=1),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
if target_name is not None:
query_parameters['targetName'] = _SERIALIZER.query("target_name", target_name, 'str')
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="HEAD",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
# fmt: on
[docs]class WebPubSubServiceClientOperationsMixin(object):
[docs] @distributed_trace
def generate_client_token(
self,
hub, # type: str
**kwargs # type: Any
):
# type: (...) -> Any
"""Generate token for the client to connect Azure Web PubSub service.
Generate token for the client to connect Azure Web PubSub service.
:param hub: Target hub name, which should start with alphabetic characters and only contain
alpha-numeric characters or underscore.
:type hub: str
:keyword user_id: User Id.
:paramtype user_id: str
:keyword role: Roles that the connection with the generated token will have.
:paramtype role: list[str]
:keyword minutes_to_expire: The expire time of the generated token.
:paramtype minutes_to_expire: int
:return: JSON object
:rtype: Any
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"token": "str" # Optional. The token value for the WebSocket client to connect to the service.
}
"""
cls = kwargs.pop('cls', None) # type: ClsType[Any]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
user_id = kwargs.pop('user_id', None) # type: Optional[str]
role = kwargs.pop('role', None) # type: Optional[List[str]]
minutes_to_expire = kwargs.pop('minutes_to_expire', 60) # type: Optional[int]
request = build_generate_client_token_request(
hub=hub,
user_id=user_id,
role=role,
minutes_to_expire=minutes_to_expire,
template_url=self.generate_client_token.metadata['url'],
)
path_format_arguments = {
"Endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(request.url, **path_format_arguments)
pipeline_response = self._client._pipeline.run(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)
if response.content:
deserialized = response.json()
else:
deserialized = None
if cls:
return cls(pipeline_response, deserialized, {})
return deserialized
generate_client_token.metadata = {'url': '/api/hubs/{hub}/:generateToken'} # type: ignore
[docs] @distributed_trace
def close_all_connections(
self,
hub, # type: str
**kwargs # type: Any
):
# type: (...) -> None
"""Close the connections in the hub.
Close the connections in the hub.
:param hub: Target hub name, which should start with alphabetic characters and only contain
alpha-numeric characters or underscore.
:type hub: str
:keyword excluded: Exclude these connectionIds when closing the connections in the hub.
:paramtype excluded: list[str]
:keyword reason: The reason closing the client connection.
:paramtype reason: str
:return: None
:rtype: None
:raises: ~azure.core.exceptions.HttpResponseError
"""
cls = kwargs.pop('cls', None) # type: ClsType[None]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
excluded = kwargs.pop('excluded', None) # type: Optional[List[str]]
reason = kwargs.pop('reason', None) # type: Optional[str]
request = build_close_all_connections_request(
hub=hub,
excluded=excluded,
reason=reason,
template_url=self.close_all_connections.metadata['url'],
)
path_format_arguments = {
"Endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(request.url, **path_format_arguments)
pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
response = pipeline_response.http_response
if response.status_code not in [204]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
raise HttpResponseError(response=response)
if cls:
return cls(pipeline_response, None, {})
close_all_connections.metadata = {'url': '/api/hubs/{hub}/:closeConnections'} # type: ignore
[docs] @distributed_trace
def send_to_all(
self,
hub, # type: str
message, # type: Union[IO, str]
**kwargs # type: Any
):
# type: (...) -> None
"""Broadcast content inside request body to all the connected client connections.
Broadcast content inside request body to all the connected client connections.
:param hub: Target hub name, which should start with alphabetic characters and only contain
alpha-numeric characters or underscore.
:type hub: str
:param message: The payload body.
:type message: IO or str
:keyword excluded: Excluded connection Ids.
:paramtype excluded: list[str]
:keyword str content_type: Media type of the body sent to the API. Default value is
"application/json". Allowed values are: "application/json", "application/octet-stream",
"text/plain."
:return: None
:rtype: None
:raises: ~azure.core.exceptions.HttpResponseError
"""
cls = kwargs.pop('cls', None) # type: ClsType[None]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
content_type = kwargs.pop('content_type', "text/plain") # type: Optional[str]
excluded = kwargs.pop('excluded', None) # type: Optional[List[str]]
json = None
content = None
if content_type.split(";")[0] in ['application/json', 'application/octet-stream']:
content = message
elif content_type.split(";")[0] in ['text/plain']:
json = message
else:
raise ValueError(
"The content_type '{}' is not one of the allowed values: "
"['application/json', 'application/octet-stream', 'text/plain']".format(content_type)
)
request = build_send_to_all_request(
hub=hub,
content_type=content_type,
excluded=excluded,
json=json,
content=content,
template_url=self.send_to_all.metadata['url'],
)
path_format_arguments = {
"Endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(request.url, **path_format_arguments)
pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
response = pipeline_response.http_response
if response.status_code not in [202]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
raise HttpResponseError(response=response)
if cls:
return cls(pipeline_response, None, {})
send_to_all.metadata = {'url': '/api/hubs/{hub}/:send'} # type: ignore
[docs] @distributed_trace
def connection_exists(
self,
hub, # type: str
connection_id, # type: str
**kwargs # type: Any
):
# type: (...) -> None
"""Check if the connection with the given connectionId exists.
Check if the connection with the given connectionId exists.
:param hub: Target hub name, which should start with alphabetic characters and only contain
alpha-numeric characters or underscore.
:type hub: str
:param connection_id: The connection Id.
:type connection_id: str
:return: None
:rtype: None
:raises: ~azure.core.exceptions.HttpResponseError
"""
cls = kwargs.pop('cls', None) # type: ClsType[None]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
request = build_connection_exists_request(
hub=hub,
connection_id=connection_id,
template_url=self.connection_exists.metadata['url'],
)
path_format_arguments = {
"Endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(request.url, **path_format_arguments)
pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
response = pipeline_response.http_response
if response.status_code not in [200, 404]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
raise HttpResponseError(response=response)
if cls:
return cls(pipeline_response, None, {})
connection_exists.metadata = {'url': '/api/hubs/{hub}/connections/{connectionId}'} # type: ignore
[docs] @distributed_trace
def close_connection(
self,
hub, # type: str
connection_id, # type: str
**kwargs # type: Any
):
# type: (...) -> None
"""Close the client connection.
Close the client connection.
:param hub: Target hub name, which should start with alphabetic characters and only contain
alpha-numeric characters or underscore.
:type hub: str
:param connection_id: Target connection Id.
:type connection_id: str
:keyword reason: The reason closing the client connection.
:paramtype reason: str
:return: None
:rtype: None
:raises: ~azure.core.exceptions.HttpResponseError
"""
cls = kwargs.pop('cls', None) # type: ClsType[None]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
reason = kwargs.pop('reason', None) # type: Optional[str]
request = build_close_connection_request(
hub=hub,
connection_id=connection_id,
reason=reason,
template_url=self.close_connection.metadata['url'],
)
path_format_arguments = {
"Endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(request.url, **path_format_arguments)
pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
response = pipeline_response.http_response
if response.status_code not in [204]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
raise HttpResponseError(response=response)
if cls:
return cls(pipeline_response, None, {})
close_connection.metadata = {'url': '/api/hubs/{hub}/connections/{connectionId}'} # type: ignore
[docs] @distributed_trace
def send_to_connection(
self,
hub, # type: str
connection_id, # type: str
message, # type: Union[IO, str]
**kwargs # type: Any
):
# type: (...) -> None
"""Send content inside request body to the specific connection.
Send content inside request body to the specific connection.
:param hub: Target hub name, which should start with alphabetic characters and only contain
alpha-numeric characters or underscore.
:type hub: str
:param connection_id: The connection Id.
:type connection_id: str
:param message: The payload body.
:type message: IO or str
:keyword str content_type: Media type of the body sent to the API. Default value is
"application/json". Allowed values are: "application/json", "application/octet-stream",
"text/plain."
:return: None
:rtype: None
:raises: ~azure.core.exceptions.HttpResponseError
"""
cls = kwargs.pop('cls', None) # type: ClsType[None]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
content_type = kwargs.pop('content_type', "text/plain") # type: Optional[str]
json = None
content = None
if content_type.split(";")[0] in ['application/json', 'application/octet-stream']:
content = message
elif content_type.split(";")[0] in ['text/plain']:
json = message
else:
raise ValueError(
"The content_type '{}' is not one of the allowed values: "
"['application/json', 'application/octet-stream', 'text/plain']".format(content_type)
)
request = build_send_to_connection_request(
hub=hub,
connection_id=connection_id,
content_type=content_type,
json=json,
content=content,
template_url=self.send_to_connection.metadata['url'],
)
path_format_arguments = {
"Endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(request.url, **path_format_arguments)
pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
response = pipeline_response.http_response
if response.status_code not in [202]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
raise HttpResponseError(response=response)
if cls:
return cls(pipeline_response, None, {})
send_to_connection.metadata = {'url': '/api/hubs/{hub}/connections/{connectionId}/:send'} # type: ignore
[docs] @distributed_trace
def group_exists(
self,
hub, # type: str
group, # type: str
**kwargs # type: Any
):
# type: (...) -> None
"""Check if there are any client connections inside the given group.
Check if there are any client connections inside the given group.
:param hub: Target hub name, which should start with alphabetic characters and only contain
alpha-numeric characters or underscore.
:type hub: str
:param group: Target group name, which length should be greater than 0 and less than 1025.
:type group: str
:return: None
:rtype: None
:raises: ~azure.core.exceptions.HttpResponseError
"""
cls = kwargs.pop('cls', None) # type: ClsType[None]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
request = build_group_exists_request(
hub=hub,
group=group,
template_url=self.group_exists.metadata['url'],
)
path_format_arguments = {
"Endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(request.url, **path_format_arguments)
pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
response = pipeline_response.http_response
if response.status_code not in [200, 404]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
raise HttpResponseError(response=response)
if cls:
return cls(pipeline_response, None, {})
group_exists.metadata = {'url': '/api/hubs/{hub}/groups/{group}'} # type: ignore
[docs] @distributed_trace
def close_group_connections(
self,
hub, # type: str
group, # type: str
**kwargs # type: Any
):
# type: (...) -> None
"""Close connections in the specific group.
Close connections in the specific group.
:param hub: Target hub name, which should start with alphabetic characters and only contain
alpha-numeric characters or underscore.
:type hub: str
:param group: Target group name, which length should be greater than 0 and less than 1025.
:type group: str
:keyword excluded: Exclude these connectionIds when closing the connections in the group.
:paramtype excluded: list[str]
:keyword reason: The reason closing the client connection.
:paramtype reason: str
:return: None
:rtype: None
:raises: ~azure.core.exceptions.HttpResponseError
"""
cls = kwargs.pop('cls', None) # type: ClsType[None]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
excluded = kwargs.pop('excluded', None) # type: Optional[List[str]]
reason = kwargs.pop('reason', None) # type: Optional[str]
request = build_close_group_connections_request(
hub=hub,
group=group,
excluded=excluded,
reason=reason,
template_url=self.close_group_connections.metadata['url'],
)
path_format_arguments = {
"Endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(request.url, **path_format_arguments)
pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
response = pipeline_response.http_response
if response.status_code not in [204]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
raise HttpResponseError(response=response)
if cls:
return cls(pipeline_response, None, {})
close_group_connections.metadata = {'url': '/api/hubs/{hub}/groups/{group}/:closeConnections'} # type: ignore
[docs] @distributed_trace
def send_to_group(
self,
hub, # type: str
group, # type: str
message, # type: Union[IO, str]
**kwargs # type: Any
):
# type: (...) -> None
"""Send content inside request body to a group of connections.
Send content inside request body to a group of connections.
:param hub: Target hub name, which should start with alphabetic characters and only contain
alpha-numeric characters or underscore.
:type hub: str
:param group: Target group name, which length should be greater than 0 and less than 1025.
:type group: str
:param message: The payload body.
:type message: IO or str
:keyword excluded: Excluded connection Ids.
:paramtype excluded: list[str]
:keyword str content_type: Media type of the body sent to the API. Default value is
"application/json". Allowed values are: "application/json", "application/octet-stream",
"text/plain."
:return: None
:rtype: None
:raises: ~azure.core.exceptions.HttpResponseError
"""
cls = kwargs.pop('cls', None) # type: ClsType[None]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
content_type = kwargs.pop('content_type', "text/plain") # type: Optional[str]
excluded = kwargs.pop('excluded', None) # type: Optional[List[str]]
json = None
content = None
if content_type.split(";")[0] in ['application/json', 'application/octet-stream']:
content = message
elif content_type.split(";")[0] in ['text/plain']:
json = message
else:
raise ValueError(
"The content_type '{}' is not one of the allowed values: "
"['application/json', 'application/octet-stream', 'text/plain']".format(content_type)
)
request = build_send_to_group_request(
hub=hub,
group=group,
content_type=content_type,
excluded=excluded,
json=json,
content=content,
template_url=self.send_to_group.metadata['url'],
)
path_format_arguments = {
"Endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(request.url, **path_format_arguments)
pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
response = pipeline_response.http_response
if response.status_code not in [202]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
raise HttpResponseError(response=response)
if cls:
return cls(pipeline_response, None, {})
send_to_group.metadata = {'url': '/api/hubs/{hub}/groups/{group}/:send'} # type: ignore
[docs] @distributed_trace
def add_connection_to_group(
self,
hub, # type: str
group, # type: str
connection_id, # type: str
**kwargs # type: Any
):
# type: (...) -> None
"""Add a connection to the target group.
Add a connection to the target group.
:param hub: Target hub name, which should start with alphabetic characters and only contain
alpha-numeric characters or underscore.
:type hub: str
:param group: Target group name, which length should be greater than 0 and less than 1025.
:type group: str
:param connection_id: Target connection Id.
:type connection_id: str
:return: None
:rtype: None
:raises: ~azure.core.exceptions.HttpResponseError
"""
cls = kwargs.pop('cls', None) # type: ClsType[None]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
request = build_add_connection_to_group_request(
hub=hub,
group=group,
connection_id=connection_id,
template_url=self.add_connection_to_group.metadata['url'],
)
path_format_arguments = {
"Endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(request.url, **path_format_arguments)
pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
response = pipeline_response.http_response
if response.status_code not in [200, 404]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
raise HttpResponseError(response=response)
if cls:
return cls(pipeline_response, None, {})
add_connection_to_group.metadata = {'url': '/api/hubs/{hub}/groups/{group}/connections/{connectionId}'} # type: ignore
[docs] @distributed_trace
def remove_connection_from_group(
self,
hub, # type: str
group, # type: str
connection_id, # type: str
**kwargs # type: Any
):
# type: (...) -> None
"""Remove a connection from the target group.
Remove a connection from the target group.
:param hub: Target hub name, which should start with alphabetic characters and only contain
alpha-numeric characters or underscore.
:type hub: str
:param group: Target group name, which length should be greater than 0 and less than 1025.
:type group: str
:param connection_id: Target connection Id.
:type connection_id: str
:return: None
:rtype: None
:raises: ~azure.core.exceptions.HttpResponseError
"""
cls = kwargs.pop('cls', None) # type: ClsType[None]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
request = build_remove_connection_from_group_request(
hub=hub,
group=group,
connection_id=connection_id,
template_url=self.remove_connection_from_group.metadata['url'],
)
path_format_arguments = {
"Endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(request.url, **path_format_arguments)
pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
response = pipeline_response.http_response
if response.status_code not in [204]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
raise HttpResponseError(response=response)
if cls:
return cls(pipeline_response, None, {})
remove_connection_from_group.metadata = {'url': '/api/hubs/{hub}/groups/{group}/connections/{connectionId}'} # type: ignore
[docs] @distributed_trace
def user_exists(
self,
hub, # type: str
user_id, # type: str
**kwargs # type: Any
):
# type: (...) -> None
"""Check if there are any client connections connected for the given user.
Check if there are any client connections connected for the given user.
:param hub: Target hub name, which should start with alphabetic characters and only contain
alpha-numeric characters or underscore.
:type hub: str
:param user_id: Target user Id.
:type user_id: str
:return: None
:rtype: None
:raises: ~azure.core.exceptions.HttpResponseError
"""
cls = kwargs.pop('cls', None) # type: ClsType[None]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
request = build_user_exists_request(
hub=hub,
user_id=user_id,
template_url=self.user_exists.metadata['url'],
)
path_format_arguments = {
"Endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(request.url, **path_format_arguments)
pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
response = pipeline_response.http_response
if response.status_code not in [200, 404]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
raise HttpResponseError(response=response)
if cls:
return cls(pipeline_response, None, {})
user_exists.metadata = {'url': '/api/hubs/{hub}/users/{userId}'} # type: ignore
[docs] @distributed_trace
def close_user_connections(
self,
hub, # type: str
user_id, # type: str
**kwargs # type: Any
):
# type: (...) -> None
"""Close connections for the specific user.
Close connections for the specific user.
:param hub: Target hub name, which should start with alphabetic characters and only contain
alpha-numeric characters or underscore.
:type hub: str
:param user_id: The user Id.
:type user_id: str
:keyword excluded: Exclude these connectionIds when closing the connections for the user.
:paramtype excluded: list[str]
:keyword reason: The reason closing the client connection.
:paramtype reason: str
:return: None
:rtype: None
:raises: ~azure.core.exceptions.HttpResponseError
"""
cls = kwargs.pop('cls', None) # type: ClsType[None]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
excluded = kwargs.pop('excluded', None) # type: Optional[List[str]]
reason = kwargs.pop('reason', None) # type: Optional[str]
request = build_close_user_connections_request(
hub=hub,
user_id=user_id,
excluded=excluded,
reason=reason,
template_url=self.close_user_connections.metadata['url'],
)
path_format_arguments = {
"Endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(request.url, **path_format_arguments)
pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
response = pipeline_response.http_response
if response.status_code not in [204]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
raise HttpResponseError(response=response)
if cls:
return cls(pipeline_response, None, {})
close_user_connections.metadata = {'url': '/api/hubs/{hub}/users/{userId}/:closeConnections'} # type: ignore
[docs] @distributed_trace
def send_to_user(
self,
hub, # type: str
user_id, # type: str
message, # type: Union[IO, str]
**kwargs # type: Any
):
# type: (...) -> None
"""Send content inside request body to the specific user.
Send content inside request body to the specific user.
:param hub: Target hub name, which should start with alphabetic characters and only contain
alpha-numeric characters or underscore.
:type hub: str
:param user_id: The user Id.
:type user_id: str
:param message: The payload body.
:type message: IO or str
:keyword str content_type: Media type of the body sent to the API. Default value is
"application/json". Allowed values are: "application/json", "application/octet-stream",
"text/plain."
:return: None
:rtype: None
:raises: ~azure.core.exceptions.HttpResponseError
"""
cls = kwargs.pop('cls', None) # type: ClsType[None]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
content_type = kwargs.pop('content_type', "text/plain") # type: Optional[str]
json = None
content = None
if content_type.split(";")[0] in ['application/json', 'application/octet-stream']:
content = message
elif content_type.split(";")[0] in ['text/plain']:
json = message
else:
raise ValueError(
"The content_type '{}' is not one of the allowed values: "
"['application/json', 'application/octet-stream', 'text/plain']".format(content_type)
)
request = build_send_to_user_request(
hub=hub,
user_id=user_id,
content_type=content_type,
json=json,
content=content,
template_url=self.send_to_user.metadata['url'],
)
path_format_arguments = {
"Endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(request.url, **path_format_arguments)
pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
response = pipeline_response.http_response
if response.status_code not in [202]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
raise HttpResponseError(response=response)
if cls:
return cls(pipeline_response, None, {})
send_to_user.metadata = {'url': '/api/hubs/{hub}/users/{userId}/:send'} # type: ignore
[docs] @distributed_trace
def add_user_to_group(
self,
hub, # type: str
group, # type: str
user_id, # type: str
**kwargs # type: Any
):
# type: (...) -> None
"""Add a user to the target group.
Add a user to the target group.
:param hub: Target hub name, which should start with alphabetic characters and only contain
alpha-numeric characters or underscore.
:type hub: str
:param group: Target group name, which length should be greater than 0 and less than 1025.
:type group: str
:param user_id: Target user Id.
:type user_id: str
:return: None
:rtype: None
:raises: ~azure.core.exceptions.HttpResponseError
"""
cls = kwargs.pop('cls', None) # type: ClsType[None]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
request = build_add_user_to_group_request(
hub=hub,
group=group,
user_id=user_id,
template_url=self.add_user_to_group.metadata['url'],
)
path_format_arguments = {
"Endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(request.url, **path_format_arguments)
pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
response = pipeline_response.http_response
if response.status_code not in [200, 404]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
raise HttpResponseError(response=response)
if cls:
return cls(pipeline_response, None, {})
add_user_to_group.metadata = {'url': '/api/hubs/{hub}/users/{userId}/groups/{group}'} # type: ignore
[docs] @distributed_trace
def remove_user_from_group(
self,
hub, # type: str
group, # type: str
user_id, # type: str
**kwargs # type: Any
):
# type: (...) -> None
"""Remove a user from the target group.
Remove a user from the target group.
:param hub: Target hub name, which should start with alphabetic characters and only contain
alpha-numeric characters or underscore.
:type hub: str
:param group: Target group name, which length should be greater than 0 and less than 1025.
:type group: str
:param user_id: Target user Id.
:type user_id: str
:return: None
:rtype: None
:raises: ~azure.core.exceptions.HttpResponseError
"""
cls = kwargs.pop('cls', None) # type: ClsType[None]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
request = build_remove_user_from_group_request(
hub=hub,
group=group,
user_id=user_id,
template_url=self.remove_user_from_group.metadata['url'],
)
path_format_arguments = {
"Endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(request.url, **path_format_arguments)
pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
response = pipeline_response.http_response
if response.status_code not in [204]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
raise HttpResponseError(response=response)
if cls:
return cls(pipeline_response, None, {})
remove_user_from_group.metadata = {'url': '/api/hubs/{hub}/users/{userId}/groups/{group}'} # type: ignore
[docs] @distributed_trace
def remove_user_from_all_groups(
self,
hub, # type: str
user_id, # type: str
**kwargs # type: Any
):
# type: (...) -> None
"""Remove a user from all groups.
Remove a user from all groups.
:param hub: Target hub name, which should start with alphabetic characters and only contain
alpha-numeric characters or underscore.
:type hub: str
:param user_id: Target user Id.
:type user_id: str
:return: None
:rtype: None
:raises: ~azure.core.exceptions.HttpResponseError
"""
cls = kwargs.pop('cls', None) # type: ClsType[None]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
request = build_remove_user_from_all_groups_request(
hub=hub,
user_id=user_id,
template_url=self.remove_user_from_all_groups.metadata['url'],
)
path_format_arguments = {
"Endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(request.url, **path_format_arguments)
pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
response = pipeline_response.http_response
if response.status_code not in [204]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
raise HttpResponseError(response=response)
if cls:
return cls(pipeline_response, None, {})
remove_user_from_all_groups.metadata = {'url': '/api/hubs/{hub}/users/{userId}/groups'} # type: ignore
[docs] @distributed_trace
def grant_permission(
self,
hub, # type: str
permission, # type: str
connection_id, # type: str
**kwargs # type: Any
):
# type: (...) -> None
"""Grant permission to the connection.
Grant permission to the connection.
:param hub: Target hub name, which should start with alphabetic characters and only contain
alpha-numeric characters or underscore.
:type hub: str
:param permission: The permission: current supported actions are joinLeaveGroup and
sendToGroup. Possible values are: "sendToGroup" or "joinLeaveGroup".
:type permission: str
:param connection_id: Target connection Id.
:type connection_id: str
:keyword target_name: The meaning of the target depends on the specific permission. For
joinLeaveGroup and sendToGroup, targetName is a required parameter standing for the group name.
:paramtype target_name: str
:return: None
:rtype: None
:raises: ~azure.core.exceptions.HttpResponseError
"""
cls = kwargs.pop('cls', None) # type: ClsType[None]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
target_name = kwargs.pop('target_name', None) # type: Optional[str]
request = build_grant_permission_request(
hub=hub,
permission=permission,
connection_id=connection_id,
target_name=target_name,
template_url=self.grant_permission.metadata['url'],
)
path_format_arguments = {
"Endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(request.url, **path_format_arguments)
pipeline_response = self._client._pipeline.run(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)
if cls:
return cls(pipeline_response, None, {})
grant_permission.metadata = {'url': '/api/hubs/{hub}/permissions/{permission}/connections/{connectionId}'} # type: ignore
[docs] @distributed_trace
def revoke_permission(
self,
hub, # type: str
permission, # type: str
connection_id, # type: str
**kwargs # type: Any
):
# type: (...) -> None
"""Revoke permission for the connection.
Revoke permission for the connection.
:param hub: Target hub name, which should start with alphabetic characters and only contain
alpha-numeric characters or underscore.
:type hub: str
:param permission: The permission: current supported actions are joinLeaveGroup and
sendToGroup. Possible values are: "sendToGroup" or "joinLeaveGroup".
:type permission: str
:param connection_id: Target connection Id.
:type connection_id: str
:keyword target_name: The meaning of the target depends on the specific permission. For
joinLeaveGroup and sendToGroup, targetName is a required parameter standing for the group name.
:paramtype target_name: str
:return: None
:rtype: None
:raises: ~azure.core.exceptions.HttpResponseError
"""
cls = kwargs.pop('cls', None) # type: ClsType[None]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
target_name = kwargs.pop('target_name', None) # type: Optional[str]
request = build_revoke_permission_request(
hub=hub,
permission=permission,
connection_id=connection_id,
target_name=target_name,
template_url=self.revoke_permission.metadata['url'],
)
path_format_arguments = {
"Endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(request.url, **path_format_arguments)
pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
response = pipeline_response.http_response
if response.status_code not in [204]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
raise HttpResponseError(response=response)
if cls:
return cls(pipeline_response, None, {})
revoke_permission.metadata = {'url': '/api/hubs/{hub}/permissions/{permission}/connections/{connectionId}'} # type: ignore
[docs] @distributed_trace
def check_permission(
self,
hub, # type: str
permission, # type: str
connection_id, # type: str
**kwargs # type: Any
):
# type: (...) -> None
"""Check if a connection has permission to the specified action.
Check if a connection has permission to the specified action.
:param hub: Target hub name, which should start with alphabetic characters and only contain
alpha-numeric characters or underscore.
:type hub: str
:param permission: The permission: current supported actions are joinLeaveGroup and
sendToGroup. Possible values are: "sendToGroup" or "joinLeaveGroup".
:type permission: str
:param connection_id: Target connection Id.
:type connection_id: str
:keyword target_name: The meaning of the target depends on the specific permission. For
joinLeaveGroup and sendToGroup, targetName is a required parameter standing for the group name.
:paramtype target_name: str
:return: None
:rtype: None
:raises: ~azure.core.exceptions.HttpResponseError
"""
cls = kwargs.pop('cls', None) # type: ClsType[None]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
target_name = kwargs.pop('target_name', None) # type: Optional[str]
request = build_check_permission_request(
hub=hub,
permission=permission,
connection_id=connection_id,
target_name=target_name,
template_url=self.check_permission.metadata['url'],
)
path_format_arguments = {
"Endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(request.url, **path_format_arguments)
pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
response = pipeline_response.http_response
if response.status_code not in [200, 404]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
raise HttpResponseError(response=response)
if cls:
return cls(pipeline_response, None, {})
check_permission.metadata = {'url': '/api/hubs/{hub}/permissions/{permission}/connections/{connectionId}'} # type: ignore