# 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. # -------------------------------------------------------------------------- # pylint: disable=line-too-long,useless-suppression,too-many-lines,name-too-long,inconsistent-return-statements from collections.abc import MutableMapping from io import IOBase import json from typing import ( Any, AsyncIterator, Callable, IO, Optional, TypeVar, Union, cast, overload, ) import urllib.parse from azure.core import AsyncPipelineClient from azure.core.async_paging import AsyncItemPaged, AsyncList from azure.core.exceptions import ( ClientAuthenticationError, HttpResponseError, ResourceExistsError, ResourceNotFoundError, ResourceNotModifiedError, StreamClosedError, StreamConsumedError, map_error, ) from azure.core.pipeline import PipelineResponse from azure.core.polling import AsyncLROPoller, AsyncNoPolling, AsyncPollingMethod from azure.core.polling.async_base_polling import AsyncLROBasePolling from azure.core.rest import AsyncHttpResponse, HttpRequest from azure.core.tracing.decorator import distributed_trace from azure.core.tracing.decorator_async import distributed_trace_async from azure.core.utils import case_insensitive_dict from ... import models as _models from ..._operations._operations import ( build_content_understanding_analyze_binary_request, build_content_understanding_analyze_request, build_content_understanding_copy_analyzer_request, build_content_understanding_create_analyzer_request, build_content_understanding_delete_analyzer_request, build_content_understanding_delete_result_request, build_content_understanding_get_analyzer_request, build_content_understanding_get_defaults_request, build_content_understanding_get_operation_status_request, build_content_understanding_get_result_file_request, build_content_understanding_get_result_request, build_content_understanding_grant_copy_authorization_request, build_content_understanding_list_analyzers_request, build_content_understanding_update_analyzer_request, build_content_understanding_update_defaults_request, ) from ..._utils.model_base import SdkJSONEncoder, _deserialize from ..._utils.utils import ClientMixinABC from .._configuration import ContentUnderstandingClientConfiguration JSON = MutableMapping[str, Any] _Unset: Any = object() T = TypeVar("T") ClsType = Optional[ Callable[[PipelineResponse[HttpRequest, AsyncHttpResponse], T, dict[str, Any]], Any] ] class _ContentUnderstandingClientOperationsMixin( ClientMixinABC[ AsyncPipelineClient[HttpRequest, AsyncHttpResponse], ContentUnderstandingClientConfiguration, ] ): async def _analyze_initial( self, analyzer_id: str, body: Union[JSON, IO[bytes]] = _Unset, *, string_encoding: str, inputs: list[_models.AnalysisInput] = _Unset, processing_location: Optional[Union[str, _models.ProcessingLocation]] = None, model_deployments: Optional[dict[str, str]] = None, **kwargs: Any ) -> AsyncIterator[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[AsyncIterator[bytes]] = kwargs.pop("cls", None) if body is _Unset: if inputs is _Unset: raise TypeError("missing required argument: inputs") body = {"inputs": inputs, "modelDeployments": model_deployments} body = {k: v for k, v in body.items() if v is not None} content_type = content_type or "application/json" _content = None if isinstance(body, (IOBase, bytes)): _content = body else: _content = json.dumps(body, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore _request = build_content_understanding_analyze_request( analyzer_id=analyzer_id, string_encoding=string_encoding, processing_location=processing_location, 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.endpoint", self._config.endpoint, "str", skip_quote=True ), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _decompress = kwargs.pop("decompress", True) _stream = True pipeline_response: PipelineResponse = await self._client._pipeline.run( # type: ignore # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [202]: try: await 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 ) raise HttpResponseError(response=response) response_headers = {} response_headers["Operation-Location"] = self._deserialize( "str", response.headers.get("Operation-Location") ) response_headers["x-ms-client-request-id"] = self._deserialize( "str", response.headers.get("x-ms-client-request-id") ) deserialized = response.iter_bytes() if _decompress else response.iter_raw() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @overload async def begin_analyze( self, analyzer_id: str, *, string_encoding: str, inputs: list[_models.AnalysisInput], processing_location: Optional[Union[str, _models.ProcessingLocation]] = None, content_type: str = "application/json", model_deployments: Optional[dict[str, str]] = None, **kwargs: Any ) -> AsyncLROPoller[_models.AnalysisResult]: """Extract content and fields from input. :param analyzer_id: The unique identifier of the analyzer. Required. :type analyzer_id: str :keyword string_encoding: The string encoding format for content spans in the response. Possible values are 'codePoint', 'utf16', and ``utf8``. Default is ``codePoint``."). Required. :paramtype string_encoding: str :keyword inputs: Inputs to analyze. Currently, only pro mode supports multiple inputs. Required. :paramtype inputs: list[~azure.ai.contentunderstanding.models.AnalysisInput] :keyword processing_location: The location where the data may be processed. Defaults to global. Known values are: "geography", "dataZone", and "global". Default value is None. :paramtype processing_location: str or ~azure.ai.contentunderstanding.models.ProcessingLocation :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :keyword model_deployments: Specify the default mapping of model names to LLM/embedding deployments in Microsoft Foundry. For details and current semantics, see `https://aka.ms/cudoc-quickstart-rest `_. Default value is None. :paramtype model_deployments: dict[str, str] :return: An instance of AsyncLROPoller that returns AnalysisResult. The AnalysisResult is compatible with MutableMapping :rtype: ~azure.core.polling.AsyncLROPoller[~azure.ai.contentunderstanding.models.AnalysisResult] :raises ~azure.core.exceptions.HttpResponseError: """ @overload async def begin_analyze( self, analyzer_id: str, body: JSON, *, string_encoding: str, processing_location: Optional[Union[str, _models.ProcessingLocation]] = None, content_type: str = "application/json", **kwargs: Any ) -> AsyncLROPoller[_models.AnalysisResult]: """Extract content and fields from input. :param analyzer_id: The unique identifier of the analyzer. Required. :type analyzer_id: str :param body: Required. :type body: JSON :keyword string_encoding: The string encoding format for content spans in the response. Possible values are 'codePoint', 'utf16', and ``utf8``. Default is ``codePoint``."). Required. :paramtype string_encoding: str :keyword processing_location: The location where the data may be processed. Defaults to global. Known values are: "geography", "dataZone", and "global". Default value is None. :paramtype processing_location: str or ~azure.ai.contentunderstanding.models.ProcessingLocation :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 AsyncLROPoller that returns AnalysisResult. The AnalysisResult is compatible with MutableMapping :rtype: ~azure.core.polling.AsyncLROPoller[~azure.ai.contentunderstanding.models.AnalysisResult] :raises ~azure.core.exceptions.HttpResponseError: """ @overload async def begin_analyze( self, analyzer_id: str, body: IO[bytes], *, string_encoding: str, processing_location: Optional[Union[str, _models.ProcessingLocation]] = None, content_type: str = "application/json", **kwargs: Any ) -> AsyncLROPoller[_models.AnalysisResult]: """Extract content and fields from input. :param analyzer_id: The unique identifier of the analyzer. Required. :type analyzer_id: str :param body: Required. :type body: IO[bytes] :keyword string_encoding: The string encoding format for content spans in the response. Possible values are 'codePoint', 'utf16', and ``utf8``. Default is ``codePoint``."). Required. :paramtype string_encoding: str :keyword processing_location: The location where the data may be processed. Defaults to global. Known values are: "geography", "dataZone", and "global". Default value is None. :paramtype processing_location: str or ~azure.ai.contentunderstanding.models.ProcessingLocation :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 AsyncLROPoller that returns AnalysisResult. The AnalysisResult is compatible with MutableMapping :rtype: ~azure.core.polling.AsyncLROPoller[~azure.ai.contentunderstanding.models.AnalysisResult] :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace_async async def begin_analyze( self, analyzer_id: str, body: Union[JSON, IO[bytes]] = _Unset, *, string_encoding: str, inputs: list[_models.AnalysisInput] = _Unset, processing_location: Optional[Union[str, _models.ProcessingLocation]] = None, model_deployments: Optional[dict[str, str]] = None, **kwargs: Any ) -> AsyncLROPoller[_models.AnalysisResult]: """Extract content and fields from input. :param analyzer_id: The unique identifier of the analyzer. Required. :type analyzer_id: str :param body: Is either a JSON type or a IO[bytes] type. Required. :type body: JSON or IO[bytes] :keyword string_encoding: The string encoding format for content spans in the response. Possible values are 'codePoint', 'utf16', and ``utf8``. Default is ``codePoint``."). Required. :paramtype string_encoding: str :keyword inputs: Inputs to analyze. Currently, only pro mode supports multiple inputs. Required. :paramtype inputs: list[~azure.ai.contentunderstanding.models.AnalysisInput] :keyword processing_location: The location where the data may be processed. Defaults to global. Known values are: "geography", "dataZone", and "global". Default value is None. :paramtype processing_location: str or ~azure.ai.contentunderstanding.models.ProcessingLocation :keyword model_deployments: Specify the default mapping of model names to LLM/embedding deployments in Microsoft Foundry. For details and current semantics, see `https://aka.ms/cudoc-quickstart-rest `_. Default value is None. :paramtype model_deployments: dict[str, str] :return: An instance of AsyncLROPoller that returns AnalysisResult. The AnalysisResult is compatible with MutableMapping :rtype: ~azure.core.polling.AsyncLROPoller[~azure.ai.contentunderstanding.models.AnalysisResult] :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.AnalysisResult] = kwargs.pop("cls", None) polling: Union[bool, AsyncPollingMethod] = 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 = await self._analyze_initial( analyzer_id=analyzer_id, body=body, string_encoding=string_encoding, inputs=inputs, processing_location=processing_location, model_deployments=model_deployments, content_type=content_type, cls=lambda x, y, z: x, headers=_headers, params=_params, **kwargs ) await raw_result.http_response.read() # type: ignore kwargs.pop("error_map", None) def get_long_running_output(pipeline_response): response_headers = {} response = pipeline_response.http_response response_headers["Operation-Location"] = self._deserialize( "str", response.headers.get("Operation-Location") ) response_headers["x-ms-client-request-id"] = self._deserialize( "str", response.headers.get("x-ms-client-request-id") ) deserialized = _deserialize( _models.AnalysisResult, response.json().get("result", {}) ) if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized path_format_arguments = { "endpoint": self._serialize.url( "self._config.endpoint", self._config.endpoint, "str", skip_quote=True ), } if polling is True: polling_method: AsyncPollingMethod = cast( AsyncPollingMethod, AsyncLROBasePolling( lro_delay, path_format_arguments=path_format_arguments, **kwargs ), ) elif polling is False: polling_method = cast(AsyncPollingMethod, AsyncNoPolling()) else: polling_method = polling if cont_token: return AsyncLROPoller[_models.AnalysisResult].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return AsyncLROPoller[_models.AnalysisResult]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) async def _analyze_binary_initial( self, analyzer_id: str, binary_input: bytes, *, string_encoding: str, content_range: Optional[str] = None, processing_location: Optional[Union[str, _models.ProcessingLocation]] = None, **kwargs: Any ) -> AsyncIterator[bytes]: 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 {} content_type: str = kwargs.pop("content_type") cls: ClsType[AsyncIterator[bytes]] = kwargs.pop("cls", None) _content = binary_input _request = build_content_understanding_analyze_binary_request( analyzer_id=analyzer_id, string_encoding=string_encoding, content_range=content_range, processing_location=processing_location, 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.endpoint", self._config.endpoint, "str", skip_quote=True ), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _decompress = kwargs.pop("decompress", True) _stream = True pipeline_response: PipelineResponse = await self._client._pipeline.run( # type: ignore # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [202]: try: await 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 ) raise HttpResponseError(response=response) response_headers = {} response_headers["Operation-Location"] = self._deserialize( "str", response.headers.get("Operation-Location") ) response_headers["x-ms-client-request-id"] = self._deserialize( "str", response.headers.get("x-ms-client-request-id") ) deserialized = response.iter_bytes() if _decompress else response.iter_raw() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @distributed_trace_async async def begin_analyze_binary( self, analyzer_id: str, binary_input: bytes, *, string_encoding: str, content_range: Optional[str] = None, processing_location: Optional[Union[str, _models.ProcessingLocation]] = None, **kwargs: Any ) -> AsyncLROPoller[_models.AnalysisResult]: """Extract content and fields from input. :param analyzer_id: The unique identifier of the analyzer. Required. :type analyzer_id: str :param binary_input: The binary content of the document to analyze. Required. :type binary_input: bytes :keyword string_encoding: The string encoding format for content spans in the response. Possible values are 'codePoint', 'utf16', and ``utf8``. Default is ``codePoint``."). Required. :paramtype string_encoding: str :keyword content_range: Range of the input to analyze (ex. ``1-3,5,9-``). Document content uses 1-based page numbers, while audio visual content uses integer milliseconds. Default value is None. :paramtype content_range: str :keyword processing_location: The location where the data may be processed. Defaults to global. Known values are: "geography", "dataZone", and "global". Default value is None. :paramtype processing_location: str or ~azure.ai.contentunderstanding.models.ProcessingLocation :return: An instance of AsyncLROPoller that returns AnalysisResult. The AnalysisResult is compatible with MutableMapping :rtype: ~azure.core.polling.AsyncLROPoller[~azure.ai.contentunderstanding.models.AnalysisResult] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} content_type: str = kwargs.pop("content_type") cls: ClsType[_models.AnalysisResult] = kwargs.pop("cls", None) polling: Union[bool, AsyncPollingMethod] = 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 = await self._analyze_binary_initial( analyzer_id=analyzer_id, binary_input=binary_input, string_encoding=string_encoding, content_range=content_range, processing_location=processing_location, content_type=content_type, cls=lambda x, y, z: x, headers=_headers, params=_params, **kwargs ) await raw_result.http_response.read() # type: ignore kwargs.pop("error_map", None) def get_long_running_output(pipeline_response): response_headers = {} response = pipeline_response.http_response response_headers["Operation-Location"] = self._deserialize( "str", response.headers.get("Operation-Location") ) response_headers["x-ms-client-request-id"] = self._deserialize( "str", response.headers.get("x-ms-client-request-id") ) deserialized = _deserialize( _models.AnalysisResult, response.json().get("result", {}) ) if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized path_format_arguments = { "endpoint": self._serialize.url( "self._config.endpoint", self._config.endpoint, "str", skip_quote=True ), } if polling is True: polling_method: AsyncPollingMethod = cast( AsyncPollingMethod, AsyncLROBasePolling( lro_delay, path_format_arguments=path_format_arguments, **kwargs ), ) elif polling is False: polling_method = cast(AsyncPollingMethod, AsyncNoPolling()) else: polling_method = polling if cont_token: return AsyncLROPoller[_models.AnalysisResult].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return AsyncLROPoller[_models.AnalysisResult]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) async def _copy_analyzer_initial( self, analyzer_id: str, body: Union[JSON, IO[bytes]] = _Unset, *, source_analyzer_id: str = _Unset, allow_replace: Optional[bool] = None, source_azure_resource_id: Optional[str] = None, source_region: Optional[str] = None, **kwargs: Any ) -> AsyncIterator[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[AsyncIterator[bytes]] = kwargs.pop("cls", None) if body is _Unset: if source_analyzer_id is _Unset: raise TypeError("missing required argument: source_analyzer_id") body = { "sourceAnalyzerId": source_analyzer_id, "sourceAzureResourceId": source_azure_resource_id, "sourceRegion": source_region, } body = {k: v for k, v in body.items() if v is not None} content_type = content_type or "application/json" _content = None if isinstance(body, (IOBase, bytes)): _content = body else: _content = json.dumps(body, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore _request = build_content_understanding_copy_analyzer_request( analyzer_id=analyzer_id, allow_replace=allow_replace, 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.endpoint", self._config.endpoint, "str", skip_quote=True ), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _decompress = kwargs.pop("decompress", True) _stream = True pipeline_response: PipelineResponse = await self._client._pipeline.run( # type: ignore # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200, 201]: try: await 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 ) raise HttpResponseError(response=response) response_headers = {} response_headers["Operation-Location"] = self._deserialize( "str", response.headers.get("Operation-Location") ) response_headers["x-ms-client-request-id"] = self._deserialize( "str", response.headers.get("x-ms-client-request-id") ) deserialized = response.iter_bytes() if _decompress else response.iter_raw() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @overload async def begin_copy_analyzer( self, analyzer_id: str, *, source_analyzer_id: str, allow_replace: Optional[bool] = None, content_type: str = "application/json", source_azure_resource_id: Optional[str] = None, source_region: Optional[str] = None, **kwargs: Any ) -> AsyncLROPoller[_models.ContentAnalyzer]: """Create a copy of the source analyzer to the current location. :param analyzer_id: The unique identifier of the analyzer. Required. :type analyzer_id: str :keyword source_analyzer_id: Source analyzer ID. Required. :paramtype source_analyzer_id: str :keyword allow_replace: Allow the operation to replace an existing resource. Default value is None. :paramtype allow_replace: bool :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :keyword source_azure_resource_id: Azure resource ID of the source analyzer location. Defaults to the current resource. Default value is None. :paramtype source_azure_resource_id: str :keyword source_region: Azure region of the source analyzer location. Defaults to current region. Default value is None. :paramtype source_region: str :return: An instance of AsyncLROPoller that returns ContentAnalyzer. The ContentAnalyzer is compatible with MutableMapping :rtype: ~azure.core.polling.AsyncLROPoller[~azure.ai.contentunderstanding.models.ContentAnalyzer] :raises ~azure.core.exceptions.HttpResponseError: """ @overload async def begin_copy_analyzer( self, analyzer_id: str, body: JSON, *, allow_replace: Optional[bool] = None, content_type: str = "application/json", **kwargs: Any ) -> AsyncLROPoller[_models.ContentAnalyzer]: """Create a copy of the source analyzer to the current location. :param analyzer_id: The unique identifier of the analyzer. Required. :type analyzer_id: str :param body: Required. :type body: JSON :keyword allow_replace: Allow the operation to replace an existing resource. Default value is None. :paramtype allow_replace: bool :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 AsyncLROPoller that returns ContentAnalyzer. The ContentAnalyzer is compatible with MutableMapping :rtype: ~azure.core.polling.AsyncLROPoller[~azure.ai.contentunderstanding.models.ContentAnalyzer] :raises ~azure.core.exceptions.HttpResponseError: """ @overload async def begin_copy_analyzer( self, analyzer_id: str, body: IO[bytes], *, allow_replace: Optional[bool] = None, content_type: str = "application/json", **kwargs: Any ) -> AsyncLROPoller[_models.ContentAnalyzer]: """Create a copy of the source analyzer to the current location. :param analyzer_id: The unique identifier of the analyzer. Required. :type analyzer_id: str :param body: Required. :type body: IO[bytes] :keyword allow_replace: Allow the operation to replace an existing resource. Default value is None. :paramtype allow_replace: bool :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 AsyncLROPoller that returns ContentAnalyzer. The ContentAnalyzer is compatible with MutableMapping :rtype: ~azure.core.polling.AsyncLROPoller[~azure.ai.contentunderstanding.models.ContentAnalyzer] :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace_async async def begin_copy_analyzer( self, analyzer_id: str, body: Union[JSON, IO[bytes]] = _Unset, *, source_analyzer_id: str = _Unset, allow_replace: Optional[bool] = None, source_azure_resource_id: Optional[str] = None, source_region: Optional[str] = None, **kwargs: Any ) -> AsyncLROPoller[_models.ContentAnalyzer]: """Create a copy of the source analyzer to the current location. :param analyzer_id: The unique identifier of the analyzer. Required. :type analyzer_id: str :param body: Is either a JSON type or a IO[bytes] type. Required. :type body: JSON or IO[bytes] :keyword source_analyzer_id: Source analyzer ID. Required. :paramtype source_analyzer_id: str :keyword allow_replace: Allow the operation to replace an existing resource. Default value is None. :paramtype allow_replace: bool :keyword source_azure_resource_id: Azure resource ID of the source analyzer location. Defaults to the current resource. Default value is None. :paramtype source_azure_resource_id: str :keyword source_region: Azure region of the source analyzer location. Defaults to current region. Default value is None. :paramtype source_region: str :return: An instance of AsyncLROPoller that returns ContentAnalyzer. The ContentAnalyzer is compatible with MutableMapping :rtype: ~azure.core.polling.AsyncLROPoller[~azure.ai.contentunderstanding.models.ContentAnalyzer] :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.ContentAnalyzer] = kwargs.pop("cls", None) polling: Union[bool, AsyncPollingMethod] = 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 = await self._copy_analyzer_initial( analyzer_id=analyzer_id, body=body, source_analyzer_id=source_analyzer_id, allow_replace=allow_replace, source_azure_resource_id=source_azure_resource_id, source_region=source_region, content_type=content_type, cls=lambda x, y, z: x, headers=_headers, params=_params, **kwargs ) await raw_result.http_response.read() # type: ignore kwargs.pop("error_map", None) def get_long_running_output(pipeline_response): response_headers = {} response = pipeline_response.http_response response_headers["Operation-Location"] = self._deserialize( "str", response.headers.get("Operation-Location") ) response_headers["x-ms-client-request-id"] = self._deserialize( "str", response.headers.get("x-ms-client-request-id") ) deserialized = _deserialize( _models.ContentAnalyzer, response.json().get("result", {}) ) if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized path_format_arguments = { "endpoint": self._serialize.url( "self._config.endpoint", self._config.endpoint, "str", skip_quote=True ), } if polling is True: polling_method: AsyncPollingMethod = cast( AsyncPollingMethod, AsyncLROBasePolling( lro_delay, path_format_arguments=path_format_arguments, **kwargs ), ) elif polling is False: polling_method = cast(AsyncPollingMethod, AsyncNoPolling()) else: polling_method = polling if cont_token: return AsyncLROPoller[_models.ContentAnalyzer].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return AsyncLROPoller[_models.ContentAnalyzer]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) async def _create_analyzer_initial( self, analyzer_id: str, resource: Union[_models.ContentAnalyzer, JSON, IO[bytes]], *, allow_replace: Optional[bool] = None, **kwargs: Any ) -> AsyncIterator[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[AsyncIterator[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_content_understanding_create_analyzer_request( analyzer_id=analyzer_id, allow_replace=allow_replace, 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.endpoint", self._config.endpoint, "str", skip_quote=True ), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _decompress = kwargs.pop("decompress", True) _stream = True pipeline_response: PipelineResponse = await self._client._pipeline.run( # type: ignore # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200, 201]: try: await 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 ) raise HttpResponseError(response=response) response_headers = {} response_headers["Operation-Location"] = self._deserialize( "str", response.headers.get("Operation-Location") ) response_headers["x-ms-client-request-id"] = self._deserialize( "str", response.headers.get("x-ms-client-request-id") ) deserialized = response.iter_bytes() if _decompress else response.iter_raw() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @overload async def begin_create_analyzer( self, analyzer_id: str, resource: _models.ContentAnalyzer, *, allow_replace: Optional[bool] = None, content_type: str = "application/json", **kwargs: Any ) -> AsyncLROPoller[_models.ContentAnalyzer]: """Create a new analyzer asynchronously. :param analyzer_id: The unique identifier of the analyzer. Required. :type analyzer_id: str :param resource: The resource instance. Required. :type resource: ~azure.ai.contentunderstanding.models.ContentAnalyzer :keyword allow_replace: Allow the operation to replace an existing resource. Default value is None. :paramtype allow_replace: bool :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 AsyncLROPoller that returns ContentAnalyzer. The ContentAnalyzer is compatible with MutableMapping :rtype: ~azure.core.polling.AsyncLROPoller[~azure.ai.contentunderstanding.models.ContentAnalyzer] :raises ~azure.core.exceptions.HttpResponseError: """ @overload async def begin_create_analyzer( self, analyzer_id: str, resource: JSON, *, allow_replace: Optional[bool] = None, content_type: str = "application/json", **kwargs: Any ) -> AsyncLROPoller[_models.ContentAnalyzer]: """Create a new analyzer asynchronously. :param analyzer_id: The unique identifier of the analyzer. Required. :type analyzer_id: str :param resource: The resource instance. Required. :type resource: JSON :keyword allow_replace: Allow the operation to replace an existing resource. Default value is None. :paramtype allow_replace: bool :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 AsyncLROPoller that returns ContentAnalyzer. The ContentAnalyzer is compatible with MutableMapping :rtype: ~azure.core.polling.AsyncLROPoller[~azure.ai.contentunderstanding.models.ContentAnalyzer] :raises ~azure.core.exceptions.HttpResponseError: """ @overload async def begin_create_analyzer( self, analyzer_id: str, resource: IO[bytes], *, allow_replace: Optional[bool] = None, content_type: str = "application/json", **kwargs: Any ) -> AsyncLROPoller[_models.ContentAnalyzer]: """Create a new analyzer asynchronously. :param analyzer_id: The unique identifier of the analyzer. Required. :type analyzer_id: str :param resource: The resource instance. Required. :type resource: IO[bytes] :keyword allow_replace: Allow the operation to replace an existing resource. Default value is None. :paramtype allow_replace: bool :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 AsyncLROPoller that returns ContentAnalyzer. The ContentAnalyzer is compatible with MutableMapping :rtype: ~azure.core.polling.AsyncLROPoller[~azure.ai.contentunderstanding.models.ContentAnalyzer] :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace_async async def begin_create_analyzer( self, analyzer_id: str, resource: Union[_models.ContentAnalyzer, JSON, IO[bytes]], *, allow_replace: Optional[bool] = None, **kwargs: Any ) -> AsyncLROPoller[_models.ContentAnalyzer]: """Create a new analyzer asynchronously. :param analyzer_id: The unique identifier of the analyzer. Required. :type analyzer_id: str :param resource: The resource instance. Is one of the following types: ContentAnalyzer, JSON, IO[bytes] Required. :type resource: ~azure.ai.contentunderstanding.models.ContentAnalyzer or JSON or IO[bytes] :keyword allow_replace: Allow the operation to replace an existing resource. Default value is None. :paramtype allow_replace: bool :return: An instance of AsyncLROPoller that returns ContentAnalyzer. The ContentAnalyzer is compatible with MutableMapping :rtype: ~azure.core.polling.AsyncLROPoller[~azure.ai.contentunderstanding.models.ContentAnalyzer] :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.ContentAnalyzer] = kwargs.pop("cls", None) polling: Union[bool, AsyncPollingMethod] = 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 = await self._create_analyzer_initial( analyzer_id=analyzer_id, resource=resource, allow_replace=allow_replace, content_type=content_type, cls=lambda x, y, z: x, headers=_headers, params=_params, **kwargs ) await raw_result.http_response.read() # type: ignore kwargs.pop("error_map", None) def get_long_running_output(pipeline_response): response_headers = {} response = pipeline_response.http_response response_headers["Operation-Location"] = self._deserialize( "str", response.headers.get("Operation-Location") ) response_headers["x-ms-client-request-id"] = self._deserialize( "str", response.headers.get("x-ms-client-request-id") ) deserialized = _deserialize(_models.ContentAnalyzer, response.json()) if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized path_format_arguments = { "endpoint": self._serialize.url( "self._config.endpoint", self._config.endpoint, "str", skip_quote=True ), } if polling is True: polling_method: AsyncPollingMethod = cast( AsyncPollingMethod, AsyncLROBasePolling( lro_delay, path_format_arguments=path_format_arguments, **kwargs ), ) elif polling is False: polling_method = cast(AsyncPollingMethod, AsyncNoPolling()) else: polling_method = polling if cont_token: return AsyncLROPoller[_models.ContentAnalyzer].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return AsyncLROPoller[_models.ContentAnalyzer]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) @distributed_trace_async async def delete_analyzer(self, analyzer_id: str, **kwargs: Any) -> None: """Delete analyzer. :param analyzer_id: The unique identifier of the analyzer. Required. :type analyzer_id: 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_content_understanding_delete_analyzer_request( analyzer_id=analyzer_id, api_version=self._config.api_version, headers=_headers, params=_params, ) 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) _stream = False pipeline_response: PipelineResponse = await self._client._pipeline.run( # type: ignore # pylint: disable=protected-access _request, stream=_stream, **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) response_headers = {} response_headers["x-ms-client-request-id"] = self._deserialize( "str", response.headers.get("x-ms-client-request-id") ) if cls: return cls(pipeline_response, None, response_headers) # type: ignore @distributed_trace_async async def delete_result(self, operation_id: str, **kwargs: Any) -> None: """Mark the result of an analysis operation for deletion. :param operation_id: Operation identifier. Required. :type operation_id: 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_content_understanding_delete_result_request( operation_id=operation_id, api_version=self._config.api_version, headers=_headers, params=_params, ) 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) _stream = False pipeline_response: PipelineResponse = await self._client._pipeline.run( # type: ignore # pylint: disable=protected-access _request, stream=_stream, **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, {}) # type: ignore @distributed_trace_async async def get_analyzer( self, analyzer_id: str, **kwargs: Any ) -> _models.ContentAnalyzer: """Get analyzer properties. :param analyzer_id: The unique identifier of the analyzer. Required. :type analyzer_id: str :return: ContentAnalyzer. The ContentAnalyzer is compatible with MutableMapping :rtype: ~azure.ai.contentunderstanding.models.ContentAnalyzer :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.ContentAnalyzer] = kwargs.pop("cls", None) _request = build_content_understanding_get_analyzer_request( analyzer_id=analyzer_id, api_version=self._config.api_version, headers=_headers, params=_params, ) 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) _decompress = kwargs.pop("decompress", True) _stream = kwargs.pop("stream", False) pipeline_response: PipelineResponse = await self._client._pipeline.run( # type: ignore # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: await 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 ) raise HttpResponseError(response=response) response_headers = {} response_headers["x-ms-client-request-id"] = self._deserialize( "str", response.headers.get("x-ms-client-request-id") ) if _stream: deserialized = response.iter_bytes() if _decompress else response.iter_raw() else: deserialized = _deserialize(_models.ContentAnalyzer, response.json()) if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @distributed_trace_async async def get_defaults(self, **kwargs: Any) -> _models.ContentUnderstandingDefaults: """Return default settings for this Content Understanding resource. :return: ContentUnderstandingDefaults. The ContentUnderstandingDefaults is compatible with MutableMapping :rtype: ~azure.ai.contentunderstanding.models.ContentUnderstandingDefaults :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.ContentUnderstandingDefaults] = kwargs.pop("cls", None) _request = build_content_understanding_get_defaults_request( api_version=self._config.api_version, headers=_headers, params=_params, ) 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) _decompress = kwargs.pop("decompress", True) _stream = kwargs.pop("stream", False) pipeline_response: PipelineResponse = await self._client._pipeline.run( # type: ignore # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: await 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 ) raise HttpResponseError(response=response) if _stream: deserialized = response.iter_bytes() if _decompress else response.iter_raw() else: deserialized = _deserialize( _models.ContentUnderstandingDefaults, response.json() ) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore @distributed_trace_async async def _get_operation_status( self, analyzer_id: str, operation_id: str, **kwargs: Any ) -> _models.ContentAnalyzerOperationStatus: """Get the status of an analyzer creation operation. :param analyzer_id: The unique identifier of the analyzer. Required. :type analyzer_id: str :param operation_id: The unique ID of the operation. Required. :type operation_id: str :return: ContentAnalyzerOperationStatus. The ContentAnalyzerOperationStatus is compatible with MutableMapping :rtype: ~azure.ai.contentunderstanding.models.ContentAnalyzerOperationStatus :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.ContentAnalyzerOperationStatus] = kwargs.pop("cls", None) _request = build_content_understanding_get_operation_status_request( analyzer_id=analyzer_id, operation_id=operation_id, api_version=self._config.api_version, headers=_headers, params=_params, ) 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) _decompress = kwargs.pop("decompress", True) _stream = kwargs.pop("stream", False) pipeline_response: PipelineResponse = await self._client._pipeline.run( # type: ignore # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: await 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 ) raise HttpResponseError(response=response) if _stream: deserialized = response.iter_bytes() if _decompress else response.iter_raw() else: deserialized = _deserialize( _models.ContentAnalyzerOperationStatus, response.json() ) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore @distributed_trace_async async def _get_result( self, operation_id: str, **kwargs: Any ) -> _models.ContentAnalyzerAnalyzeOperationStatus: """Get the result of an analysis operation. :param operation_id: The unique ID of the operation. Required. :type operation_id: str :return: ContentAnalyzerAnalyzeOperationStatus. The ContentAnalyzerAnalyzeOperationStatus is compatible with MutableMapping :rtype: ~azure.ai.contentunderstanding.models.ContentAnalyzerAnalyzeOperationStatus :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.ContentAnalyzerAnalyzeOperationStatus] = kwargs.pop( "cls", None ) _request = build_content_understanding_get_result_request( operation_id=operation_id, api_version=self._config.api_version, headers=_headers, params=_params, ) 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) _decompress = kwargs.pop("decompress", True) _stream = kwargs.pop("stream", False) pipeline_response: PipelineResponse = await self._client._pipeline.run( # type: ignore # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: await 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 ) raise HttpResponseError(response=response) if _stream: deserialized = response.iter_bytes() if _decompress else response.iter_raw() else: deserialized = _deserialize( _models.ContentAnalyzerAnalyzeOperationStatus, response.json() ) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore @distributed_trace_async async def get_result_file( self, operation_id: str, path: str, **kwargs: Any ) -> AsyncIterator[bytes]: """Get a file associated with the result of an analysis operation. :param operation_id: Operation identifier. Required. :type operation_id: str :param path: File path. Required. :type path: str :return: AsyncIterator[bytes] :rtype: AsyncIterator[bytes] :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[AsyncIterator[bytes]] = kwargs.pop("cls", None) _request = build_content_understanding_get_result_file_request( operation_id=operation_id, path=path, api_version=self._config.api_version, headers=_headers, params=_params, ) 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) _decompress = kwargs.pop("decompress", True) _stream = kwargs.pop("stream", True) pipeline_response: PipelineResponse = await self._client._pipeline.run( # type: ignore # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: await 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 ) raise HttpResponseError(response=response) response_headers = {} response_headers["content-type"] = self._deserialize( "str", response.headers.get("content-type") ) deserialized = response.iter_bytes() if _decompress else response.iter_raw() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @overload async def grant_copy_authorization( self, analyzer_id: str, *, target_azure_resource_id: str, content_type: str = "application/json", target_region: Optional[str] = None, **kwargs: Any ) -> _models.CopyAuthorization: """Get authorization for copying this analyzer to another location. :param analyzer_id: The unique identifier of the analyzer. Required. :type analyzer_id: str :keyword target_azure_resource_id: Azure resource ID of the target analyzer location. Required. :paramtype target_azure_resource_id: str :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :keyword target_region: Azure region of the target analyzer location. Defaults to current region. Default value is None. :paramtype target_region: str :return: CopyAuthorization. The CopyAuthorization is compatible with MutableMapping :rtype: ~azure.ai.contentunderstanding.models.CopyAuthorization :raises ~azure.core.exceptions.HttpResponseError: """ @overload async def grant_copy_authorization( self, analyzer_id: str, body: JSON, *, content_type: str = "application/json", **kwargs: Any ) -> _models.CopyAuthorization: """Get authorization for copying this analyzer to another location. :param analyzer_id: The unique identifier of the analyzer. Required. :type analyzer_id: str :param body: Required. :type body: JSON :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :return: CopyAuthorization. The CopyAuthorization is compatible with MutableMapping :rtype: ~azure.ai.contentunderstanding.models.CopyAuthorization :raises ~azure.core.exceptions.HttpResponseError: """ @overload async def grant_copy_authorization( self, analyzer_id: str, body: IO[bytes], *, content_type: str = "application/json", **kwargs: Any ) -> _models.CopyAuthorization: """Get authorization for copying this analyzer to another location. :param analyzer_id: The unique identifier of the analyzer. Required. :type analyzer_id: str :param body: Required. :type body: 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: CopyAuthorization. The CopyAuthorization is compatible with MutableMapping :rtype: ~azure.ai.contentunderstanding.models.CopyAuthorization :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace_async async def grant_copy_authorization( self, analyzer_id: str, body: Union[JSON, IO[bytes]] = _Unset, *, target_azure_resource_id: str = _Unset, target_region: Optional[str] = None, **kwargs: Any ) -> _models.CopyAuthorization: """Get authorization for copying this analyzer to another location. :param analyzer_id: The unique identifier of the analyzer. Required. :type analyzer_id: str :param body: Is either a JSON type or a IO[bytes] type. Required. :type body: JSON or IO[bytes] :keyword target_azure_resource_id: Azure resource ID of the target analyzer location. Required. :paramtype target_azure_resource_id: str :keyword target_region: Azure region of the target analyzer location. Defaults to current region. Default value is None. :paramtype target_region: str :return: CopyAuthorization. The CopyAuthorization is compatible with MutableMapping :rtype: ~azure.ai.contentunderstanding.models.CopyAuthorization :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.CopyAuthorization] = kwargs.pop("cls", None) if body is _Unset: if target_azure_resource_id is _Unset: raise TypeError("missing required argument: target_azure_resource_id") body = { "targetAzureResourceId": target_azure_resource_id, "targetRegion": target_region, } body = {k: v for k, v in body.items() if v is not None} content_type = content_type or "application/json" _content = None if isinstance(body, (IOBase, bytes)): _content = body else: _content = json.dumps(body, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore _request = build_content_understanding_grant_copy_authorization_request( analyzer_id=analyzer_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.endpoint", self._config.endpoint, "str", skip_quote=True ), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _decompress = kwargs.pop("decompress", True) _stream = kwargs.pop("stream", False) pipeline_response: PipelineResponse = await self._client._pipeline.run( # type: ignore # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: await 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 ) raise HttpResponseError(response=response) response_headers = {} response_headers["x-ms-client-request-id"] = self._deserialize( "str", response.headers.get("x-ms-client-request-id") ) if _stream: deserialized = response.iter_bytes() if _decompress else response.iter_raw() else: deserialized = _deserialize(_models.CopyAuthorization, response.json()) if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @distributed_trace def list_analyzers( self, **kwargs: Any ) -> AsyncItemPaged["_models.ContentAnalyzer"]: """List analyzers. :return: An iterator like instance of ContentAnalyzer :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.ai.contentunderstanding.models.ContentAnalyzer] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[list[_models.ContentAnalyzer]] = 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_content_understanding_list_analyzers_request( api_version=self._config.api_version, headers=_headers, params=_params, ) 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 ) 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.endpoint", self._config.endpoint, "str", skip_quote=True, ), } _request.url = self._client.format_url( _request.url, **path_format_arguments ) return _request async def extract_data(pipeline_response): deserialized = pipeline_response.http_response.json() list_of_elem = _deserialize( list[_models.ContentAnalyzer], deserialized.get("value", []), ) if cls: list_of_elem = cls(list_of_elem) # type: ignore return deserialized.get("nextLink") or None, AsyncList(list_of_elem) async def get_next(next_link=None): _request = prepare_request(next_link) _stream = False pipeline_response: PipelineResponse = await self._client._pipeline.run( # type: ignore # 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, ) raise HttpResponseError(response=response) return pipeline_response return AsyncItemPaged(get_next, extract_data) @overload async def update_analyzer( self, analyzer_id: str, resource: _models.ContentAnalyzer, *, content_type: str = "application/merge-patch+json", **kwargs: Any ) -> _models.ContentAnalyzer: """Update analyzer properties. :param analyzer_id: The unique identifier of the analyzer. Required. :type analyzer_id: str :param resource: The resource instance. Required. :type resource: ~azure.ai.contentunderstanding.models.ContentAnalyzer :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/merge-patch+json". :paramtype content_type: str :return: ContentAnalyzer. The ContentAnalyzer is compatible with MutableMapping :rtype: ~azure.ai.contentunderstanding.models.ContentAnalyzer :raises ~azure.core.exceptions.HttpResponseError: """ @overload async def update_analyzer( self, analyzer_id: str, resource: JSON, *, content_type: str = "application/merge-patch+json", **kwargs: Any ) -> _models.ContentAnalyzer: """Update analyzer properties. :param analyzer_id: The unique identifier of the analyzer. Required. :type analyzer_id: str :param resource: The resource instance. Required. :type resource: JSON :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/merge-patch+json". :paramtype content_type: str :return: ContentAnalyzer. The ContentAnalyzer is compatible with MutableMapping :rtype: ~azure.ai.contentunderstanding.models.ContentAnalyzer :raises ~azure.core.exceptions.HttpResponseError: """ @overload async def update_analyzer( self, analyzer_id: str, resource: IO[bytes], *, content_type: str = "application/merge-patch+json", **kwargs: Any ) -> _models.ContentAnalyzer: """Update analyzer properties. :param analyzer_id: The unique identifier of the analyzer. Required. :type analyzer_id: str :param resource: The resource instance. Required. :type resource: IO[bytes] :keyword content_type: Body Parameter content-type. Content type parameter for binary body. Default value is "application/merge-patch+json". :paramtype content_type: str :return: ContentAnalyzer. The ContentAnalyzer is compatible with MutableMapping :rtype: ~azure.ai.contentunderstanding.models.ContentAnalyzer :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace_async async def update_analyzer( self, analyzer_id: str, resource: Union[_models.ContentAnalyzer, JSON, IO[bytes]], **kwargs: Any ) -> _models.ContentAnalyzer: """Update analyzer properties. :param analyzer_id: The unique identifier of the analyzer. Required. :type analyzer_id: str :param resource: The resource instance. Is one of the following types: ContentAnalyzer, JSON, IO[bytes] Required. :type resource: ~azure.ai.contentunderstanding.models.ContentAnalyzer or JSON or IO[bytes] :return: ContentAnalyzer. The ContentAnalyzer is compatible with MutableMapping :rtype: ~azure.ai.contentunderstanding.models.ContentAnalyzer :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.ContentAnalyzer] = kwargs.pop("cls", None) content_type = content_type or "application/merge-patch+json" _content = None if isinstance(resource, (IOBase, bytes)): _content = resource else: _content = json.dumps(resource, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore _request = build_content_understanding_update_analyzer_request( analyzer_id=analyzer_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.endpoint", self._config.endpoint, "str", skip_quote=True ), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _decompress = kwargs.pop("decompress", True) _stream = kwargs.pop("stream", False) pipeline_response: PipelineResponse = await self._client._pipeline.run( # type: ignore # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: await 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 ) raise HttpResponseError(response=response) response_headers = {} response_headers["x-ms-client-request-id"] = self._deserialize( "str", response.headers.get("x-ms-client-request-id") ) if _stream: deserialized = response.iter_bytes() if _decompress else response.iter_raw() else: deserialized = _deserialize(_models.ContentAnalyzer, response.json()) if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @overload async def update_defaults( self, *, content_type: str = "application/merge-patch+json", model_deployments: Optional[_models.RecordMergePatchUpdate] = None, **kwargs: Any ) -> _models.ContentUnderstandingDefaults: """Update default settings for this Content Understanding resource. :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/merge-patch+json". :paramtype content_type: str :keyword model_deployments: Specify the default mapping of model names to LLM/embedding deployments in Microsoft Foundry. For details and current semantics, see `https://aka.ms/cudoc-quickstart-rest `_. Default value is None. :paramtype model_deployments: ~azure.ai.contentunderstanding.models.RecordMergePatchUpdate :return: ContentUnderstandingDefaults. The ContentUnderstandingDefaults is compatible with MutableMapping :rtype: ~azure.ai.contentunderstanding.models.ContentUnderstandingDefaults :raises ~azure.core.exceptions.HttpResponseError: """ @overload async def update_defaults( self, body: JSON, *, content_type: str = "application/merge-patch+json", **kwargs: Any ) -> _models.ContentUnderstandingDefaults: """Update default settings for this Content Understanding resource. :param body: Required. :type body: JSON :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/merge-patch+json". :paramtype content_type: str :return: ContentUnderstandingDefaults. The ContentUnderstandingDefaults is compatible with MutableMapping :rtype: ~azure.ai.contentunderstanding.models.ContentUnderstandingDefaults :raises ~azure.core.exceptions.HttpResponseError: """ @overload async def update_defaults( self, body: IO[bytes], *, content_type: str = "application/merge-patch+json", **kwargs: Any ) -> _models.ContentUnderstandingDefaults: """Update default settings for this Content Understanding resource. :param body: Required. :type body: IO[bytes] :keyword content_type: Body Parameter content-type. Content type parameter for binary body. Default value is "application/merge-patch+json". :paramtype content_type: str :return: ContentUnderstandingDefaults. The ContentUnderstandingDefaults is compatible with MutableMapping :rtype: ~azure.ai.contentunderstanding.models.ContentUnderstandingDefaults :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace_async async def update_defaults( self, body: Union[JSON, IO[bytes]] = _Unset, *, model_deployments: Optional[_models.RecordMergePatchUpdate] = None, **kwargs: Any ) -> _models.ContentUnderstandingDefaults: """Update default settings for this Content Understanding resource. :param body: Is either a JSON type or a IO[bytes] type. Required. :type body: JSON or IO[bytes] :keyword model_deployments: Specify the default mapping of model names to LLM/embedding deployments in Microsoft Foundry. For details and current semantics, see `https://aka.ms/cudoc-quickstart-rest `_. Default value is None. :paramtype model_deployments: ~azure.ai.contentunderstanding.models.RecordMergePatchUpdate :return: ContentUnderstandingDefaults. The ContentUnderstandingDefaults is compatible with MutableMapping :rtype: ~azure.ai.contentunderstanding.models.ContentUnderstandingDefaults :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.ContentUnderstandingDefaults] = kwargs.pop("cls", None) if body is _Unset: body = {"modelDeployments": model_deployments} body = {k: v for k, v in body.items() if v is not None} content_type = content_type or "application/merge-patch+json" _content = None if isinstance(body, (IOBase, bytes)): _content = body else: _content = json.dumps(body, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore _request = build_content_understanding_update_defaults_request( 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.endpoint", self._config.endpoint, "str", skip_quote=True ), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _decompress = kwargs.pop("decompress", True) _stream = kwargs.pop("stream", False) pipeline_response: PipelineResponse = await self._client._pipeline.run( # type: ignore # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: await 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 ) raise HttpResponseError(response=response) if _stream: deserialized = response.iter_bytes() if _decompress else response.iter_raw() else: deserialized = _deserialize( _models.ContentUnderstandingDefaults, response.json() ) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore