# 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=too-many-lines,name-too-long,inconsistent-return-statements from collections.abc import MutableMapping from io import IOBase import json from typing import Any, Callable, IO, Iterator, Optional, TypeVar, Union, cast, overload import urllib.parse from azure.core import PipelineClient from azure.core.exceptions import ( ClientAuthenticationError, HttpResponseError, ResourceExistsError, ResourceNotFoundError, ResourceNotModifiedError, StreamClosedError, StreamConsumedError, map_error, ) from azure.core.paging import ItemPaged from azure.core.pipeline import PipelineResponse from azure.core.polling import LROPoller, NoPolling, PollingMethod from azure.core.polling.base_polling import LROBasePolling from azure.core.rest import HttpRequest, HttpResponse from azure.core.tracing.decorator import distributed_trace from azure.core.utils import case_insensitive_dict from .. import models as _models from .._configuration import ContentUnderstandingClientConfiguration from .._utils.model_base import SdkJSONEncoder, _deserialize from .._utils.serialization import Serializer from .._utils.utils import ClientMixinABC JSON = MutableMapping[str, Any] _Unset: Any = object() T = TypeVar("T") ClsType = Optional[ Callable[[PipelineResponse[HttpRequest, HttpResponse], T, dict[str, Any]], Any] ] _SERIALIZER = Serializer() _SERIALIZER.client_side_validation = False def build_content_understanding_analyze_request( # pylint: disable=name-too-long analyzer_id: str, *, string_encoding: str, processing_location: Optional[Union[str, _models.ProcessingLocation]] = None, **kwargs: Any, ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) content_type: Optional[str] = kwargs.pop( "content_type", _headers.pop("Content-Type", None) ) api_version: str = kwargs.pop( "api_version", _params.pop("api-version", "2025-11-01") ) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/analyzers/{analyzerId}:analyze" path_format_arguments = { "analyzerId": _SERIALIZER.url("analyzer_id", analyzer_id, "str"), } _url: str = _url.format(**path_format_arguments) # type: ignore # Construct parameters _params["stringEncoding"] = _SERIALIZER.query( "string_encoding", string_encoding, "str" ) if processing_location is not None: _params["processingLocation"] = _SERIALIZER.query( "processing_location", processing_location, "str" ) _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str") # Construct headers if content_type is not None: _headers["Content-Type"] = _SERIALIZER.header( "content_type", content_type, "str" ) _headers["Accept"] = _SERIALIZER.header("accept", accept, "str") return HttpRequest( method="POST", url=_url, params=_params, headers=_headers, **kwargs ) def build_content_understanding_analyze_binary_request( # pylint: disable=name-too-long analyzer_id: str, *, string_encoding: str, content_range: Optional[str] = None, processing_location: Optional[Union[str, _models.ProcessingLocation]] = None, **kwargs: Any, ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) content_type: str = kwargs.pop("content_type") api_version: str = kwargs.pop( "api_version", _params.pop("api-version", "2025-11-01") ) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/analyzers/{analyzerId}:analyzeBinary" path_format_arguments = { "analyzerId": _SERIALIZER.url("analyzer_id", analyzer_id, "str"), } _url: str = _url.format(**path_format_arguments) # type: ignore # Construct parameters _params["stringEncoding"] = _SERIALIZER.query( "string_encoding", string_encoding, "str" ) if content_range is not None: _params["range"] = _SERIALIZER.query("content_range", content_range, "str") if processing_location is not None: _params["processingLocation"] = _SERIALIZER.query( "processing_location", processing_location, "str" ) _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str") # Construct headers _headers["content-type"] = _SERIALIZER.header("content_type", content_type, "str") _headers["Accept"] = _SERIALIZER.header("accept", accept, "str") return HttpRequest( method="POST", url=_url, params=_params, headers=_headers, **kwargs ) def build_content_understanding_copy_analyzer_request( # pylint: disable=name-too-long analyzer_id: str, *, allow_replace: Optional[bool] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) content_type: Optional[str] = kwargs.pop( "content_type", _headers.pop("Content-Type", None) ) api_version: str = kwargs.pop( "api_version", _params.pop("api-version", "2025-11-01") ) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/analyzers/{analyzerId}:copy" path_format_arguments = { "analyzerId": _SERIALIZER.url("analyzer_id", analyzer_id, "str"), } _url: str = _url.format(**path_format_arguments) # type: ignore # Construct parameters _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str") if allow_replace is not None: _params["allowReplace"] = _SERIALIZER.query( "allow_replace", allow_replace, "bool" ) # Construct headers if content_type is not None: _headers["Content-Type"] = _SERIALIZER.header( "content_type", content_type, "str" ) _headers["Accept"] = _SERIALIZER.header("accept", accept, "str") return HttpRequest( method="POST", url=_url, params=_params, headers=_headers, **kwargs ) def build_content_understanding_create_analyzer_request( # pylint: disable=name-too-long analyzer_id: str, *, allow_replace: Optional[bool] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) content_type: Optional[str] = kwargs.pop( "content_type", _headers.pop("Content-Type", None) ) api_version: str = kwargs.pop( "api_version", _params.pop("api-version", "2025-11-01") ) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/analyzers/{analyzerId}" path_format_arguments = { "analyzerId": _SERIALIZER.url("analyzer_id", analyzer_id, "str"), } _url: str = _url.format(**path_format_arguments) # type: ignore # Construct parameters _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str") if allow_replace is not None: _params["allowReplace"] = _SERIALIZER.query( "allow_replace", allow_replace, "bool" ) # Construct headers if content_type is not None: _headers["Content-Type"] = _SERIALIZER.header( "content_type", content_type, "str" ) _headers["Accept"] = _SERIALIZER.header("accept", accept, "str") return HttpRequest( method="PUT", url=_url, params=_params, headers=_headers, **kwargs ) def build_content_understanding_delete_analyzer_request( # pylint: disable=name-too-long analyzer_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop( "api_version", _params.pop("api-version", "2025-11-01") ) # Construct URL _url = "/analyzers/{analyzerId}" path_format_arguments = { "analyzerId": _SERIALIZER.url("analyzer_id", analyzer_id, "str"), } _url: str = _url.format(**path_format_arguments) # type: ignore # Construct parameters _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str") # Construct headers return HttpRequest( method="DELETE", url=_url, params=_params, headers=_headers, **kwargs ) def build_content_understanding_delete_result_request( # pylint: disable=name-too-long operation_id: str, **kwargs: Any ) -> HttpRequest: _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop( "api_version", _params.pop("api-version", "2025-11-01") ) # Construct URL _url = "/analyzerResults/{operationId}" path_format_arguments = { "operationId": _SERIALIZER.url("operation_id", operation_id, "str"), } _url: str = _url.format(**path_format_arguments) # type: ignore # Construct parameters _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str") return HttpRequest(method="DELETE", url=_url, params=_params, **kwargs) def build_content_understanding_get_analyzer_request( # pylint: disable=name-too-long analyzer_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop( "api_version", _params.pop("api-version", "2025-11-01") ) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/analyzers/{analyzerId}" path_format_arguments = { "analyzerId": _SERIALIZER.url("analyzer_id", analyzer_id, "str"), } _url: str = _url.format(**path_format_arguments) # type: ignore # Construct parameters _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str") # Construct headers _headers["Accept"] = _SERIALIZER.header("accept", accept, "str") return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_content_understanding_get_defaults_request( **kwargs: Any, ) -> HttpRequest: # pylint: disable=name-too-long _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop( "api_version", _params.pop("api-version", "2025-11-01") ) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/defaults" # Construct parameters _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str") # Construct headers _headers["Accept"] = _SERIALIZER.header("accept", accept, "str") return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_content_understanding_get_operation_status_request( # pylint: disable=name-too-long analyzer_id: str, operation_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop( "api_version", _params.pop("api-version", "2025-11-01") ) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/analyzers/{analyzerId}/operations/{operationId}" path_format_arguments = { "analyzerId": _SERIALIZER.url("analyzer_id", analyzer_id, "str"), "operationId": _SERIALIZER.url("operation_id", operation_id, "str"), } _url: str = _url.format(**path_format_arguments) # type: ignore # Construct parameters _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str") # Construct headers _headers["Accept"] = _SERIALIZER.header("accept", accept, "str") return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_content_understanding_get_result_request( # pylint: disable=name-too-long operation_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop( "api_version", _params.pop("api-version", "2025-11-01") ) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/analyzerResults/{operationId}" path_format_arguments = { "operationId": _SERIALIZER.url("operation_id", operation_id, "str"), } _url: str = _url.format(**path_format_arguments) # type: ignore # Construct parameters _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str") # Construct headers _headers["Accept"] = _SERIALIZER.header("accept", accept, "str") return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_content_understanding_get_result_file_request( # pylint: disable=name-too-long operation_id: str, path: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop( "api_version", _params.pop("api-version", "2025-11-01") ) accept = _headers.pop("Accept", "*/*") # Construct URL _url = "/analyzerResults/{operationId}/files/{path}" path_format_arguments = { "operationId": _SERIALIZER.url("operation_id", operation_id, "str"), "path": _SERIALIZER.url("path", path, "str", skip_quote=True), } _url: str = _url.format(**path_format_arguments) # type: ignore # Construct parameters _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str") # Construct headers _headers["Accept"] = _SERIALIZER.header("accept", accept, "str") return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_content_understanding_grant_copy_authorization_request( # pylint: disable=name-too-long analyzer_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) content_type: Optional[str] = kwargs.pop( "content_type", _headers.pop("Content-Type", None) ) api_version: str = kwargs.pop( "api_version", _params.pop("api-version", "2025-11-01") ) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/analyzers/{analyzerId}:grantCopyAuthorization" path_format_arguments = { "analyzerId": _SERIALIZER.url("analyzer_id", analyzer_id, "str"), } _url: str = _url.format(**path_format_arguments) # type: ignore # Construct parameters _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str") # Construct headers if content_type is not None: _headers["Content-Type"] = _SERIALIZER.header( "content_type", content_type, "str" ) _headers["Accept"] = _SERIALIZER.header("accept", accept, "str") return HttpRequest( method="POST", url=_url, params=_params, headers=_headers, **kwargs ) def build_content_understanding_list_analyzers_request( **kwargs: Any, ) -> HttpRequest: # pylint: disable=name-too-long _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop( "api_version", _params.pop("api-version", "2025-11-01") ) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/analyzers" # Construct parameters _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str") # Construct headers _headers["Accept"] = _SERIALIZER.header("accept", accept, "str") return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_content_understanding_update_analyzer_request( # pylint: disable=name-too-long analyzer_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) content_type: Optional[str] = kwargs.pop( "content_type", _headers.pop("Content-Type", None) ) api_version: str = kwargs.pop( "api_version", _params.pop("api-version", "2025-11-01") ) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/analyzers/{analyzerId}" path_format_arguments = { "analyzerId": _SERIALIZER.url("analyzer_id", analyzer_id, "str"), } _url: str = _url.format(**path_format_arguments) # type: ignore # Construct parameters _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str") # Construct headers if content_type is not None: _headers["Content-Type"] = _SERIALIZER.header( "content_type", content_type, "str" ) _headers["Accept"] = _SERIALIZER.header("accept", accept, "str") return HttpRequest( method="PATCH", url=_url, params=_params, headers=_headers, **kwargs ) def build_content_understanding_update_defaults_request( **kwargs: Any, ) -> HttpRequest: # pylint: disable=name-too-long _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) content_type: Optional[str] = kwargs.pop( "content_type", _headers.pop("Content-Type", None) ) api_version: str = kwargs.pop( "api_version", _params.pop("api-version", "2025-11-01") ) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/defaults" # Construct parameters _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str") # Construct headers if content_type is not None: _headers["Content-Type"] = _SERIALIZER.header( "content_type", content_type, "str" ) _headers["Accept"] = _SERIALIZER.header("accept", accept, "str") return HttpRequest( method="PATCH", url=_url, params=_params, headers=_headers, **kwargs ) class _ContentUnderstandingClientOperationsMixin( ClientMixinABC[ PipelineClient[HttpRequest, HttpResponse], ContentUnderstandingClientConfiguration, ] ): 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, ) -> Iterator[bytes]: error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } error_map.update(kwargs.pop("error_map", {}) or {}) _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = kwargs.pop("params", {}) or {} content_type: Optional[str] = kwargs.pop( "content_type", _headers.pop("Content-Type", None) ) cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) 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 = ( self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) ) response = pipeline_response.http_response if response.status_code not in [202]: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error( status_code=response.status_code, response=response, error_map=error_map ) 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 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, ) -> LROPoller[_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 LROPoller that returns AnalysisResult. The AnalysisResult is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.contentunderstanding.models.AnalysisResult] :raises ~azure.core.exceptions.HttpResponseError: """ @overload 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, ) -> LROPoller[_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 LROPoller that returns AnalysisResult. The AnalysisResult is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.contentunderstanding.models.AnalysisResult] :raises ~azure.core.exceptions.HttpResponseError: """ @overload 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, ) -> LROPoller[_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 LROPoller that returns AnalysisResult. The AnalysisResult is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.contentunderstanding.models.AnalysisResult] :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace 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, ) -> LROPoller[_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 LROPoller that returns AnalysisResult. The AnalysisResult is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~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, PollingMethod] = kwargs.pop("polling", True) lro_delay = kwargs.pop("polling_interval", self._config.polling_interval) cont_token: Optional[str] = kwargs.pop("continuation_token", None) if cont_token is None: raw_result = self._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, ) 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: PollingMethod = cast( PollingMethod, LROBasePolling( lro_delay, path_format_arguments=path_format_arguments, **kwargs ), ) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.AnalysisResult].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.AnalysisResult]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) 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, ) -> Iterator[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[Iterator[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 = ( self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) ) response = pipeline_response.http_response if response.status_code not in [202]: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error( status_code=response.status_code, response=response, error_map=error_map ) 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 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, ) -> LROPoller[_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 LROPoller that returns AnalysisResult. The AnalysisResult is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~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, PollingMethod] = kwargs.pop("polling", True) lro_delay = kwargs.pop("polling_interval", self._config.polling_interval) cont_token: Optional[str] = kwargs.pop("continuation_token", None) if cont_token is None: raw_result = self._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, ) 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: PollingMethod = cast( PollingMethod, LROBasePolling( lro_delay, path_format_arguments=path_format_arguments, **kwargs ), ) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.AnalysisResult].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.AnalysisResult]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) 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, ) -> Iterator[bytes]: error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } error_map.update(kwargs.pop("error_map", {}) or {}) _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = kwargs.pop("params", {}) or {} content_type: Optional[str] = kwargs.pop( "content_type", _headers.pop("Content-Type", None) ) cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) 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 = ( self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) ) response = pipeline_response.http_response if response.status_code not in [200, 201]: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error( status_code=response.status_code, response=response, error_map=error_map ) 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 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, ) -> LROPoller[_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 LROPoller that returns ContentAnalyzer. The ContentAnalyzer is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.contentunderstanding.models.ContentAnalyzer] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_copy_analyzer( self, analyzer_id: str, body: JSON, *, allow_replace: Optional[bool] = None, content_type: str = "application/json", **kwargs: Any, ) -> LROPoller[_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 LROPoller that returns ContentAnalyzer. The ContentAnalyzer is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.contentunderstanding.models.ContentAnalyzer] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_copy_analyzer( self, analyzer_id: str, body: IO[bytes], *, allow_replace: Optional[bool] = None, content_type: str = "application/json", **kwargs: Any, ) -> LROPoller[_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 LROPoller that returns ContentAnalyzer. The ContentAnalyzer is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.contentunderstanding.models.ContentAnalyzer] :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace 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, ) -> LROPoller[_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 LROPoller that returns ContentAnalyzer. The ContentAnalyzer is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~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, PollingMethod] = kwargs.pop("polling", True) lro_delay = kwargs.pop("polling_interval", self._config.polling_interval) cont_token: Optional[str] = kwargs.pop("continuation_token", None) if cont_token is None: raw_result = self._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, ) 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: PollingMethod = cast( PollingMethod, LROBasePolling( lro_delay, path_format_arguments=path_format_arguments, **kwargs ), ) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.ContentAnalyzer].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.ContentAnalyzer]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) def _create_analyzer_initial( self, analyzer_id: str, resource: Union[_models.ContentAnalyzer, JSON, IO[bytes]], *, allow_replace: Optional[bool] = None, **kwargs: Any, ) -> Iterator[bytes]: error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } error_map.update(kwargs.pop("error_map", {}) or {}) _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = kwargs.pop("params", {}) or {} content_type: Optional[str] = kwargs.pop( "content_type", _headers.pop("Content-Type", None) ) cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) content_type = content_type or "application/json" _content = None if isinstance(resource, (IOBase, bytes)): _content = resource else: _content = json.dumps(resource, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore _request = build_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 = ( self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) ) response = pipeline_response.http_response if response.status_code not in [200, 201]: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error( status_code=response.status_code, response=response, error_map=error_map ) 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 def begin_create_analyzer( self, analyzer_id: str, resource: _models.ContentAnalyzer, *, allow_replace: Optional[bool] = None, content_type: str = "application/json", **kwargs: Any, ) -> LROPoller[_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 LROPoller that returns ContentAnalyzer. The ContentAnalyzer is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.contentunderstanding.models.ContentAnalyzer] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_create_analyzer( self, analyzer_id: str, resource: JSON, *, allow_replace: Optional[bool] = None, content_type: str = "application/json", **kwargs: Any, ) -> LROPoller[_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 LROPoller that returns ContentAnalyzer. The ContentAnalyzer is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.contentunderstanding.models.ContentAnalyzer] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_create_analyzer( self, analyzer_id: str, resource: IO[bytes], *, allow_replace: Optional[bool] = None, content_type: str = "application/json", **kwargs: Any, ) -> LROPoller[_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 LROPoller that returns ContentAnalyzer. The ContentAnalyzer is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.contentunderstanding.models.ContentAnalyzer] :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace def begin_create_analyzer( self, analyzer_id: str, resource: Union[_models.ContentAnalyzer, JSON, IO[bytes]], *, allow_replace: Optional[bool] = None, **kwargs: Any, ) -> LROPoller[_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 LROPoller that returns ContentAnalyzer. The ContentAnalyzer is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~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, PollingMethod] = kwargs.pop("polling", True) lro_delay = kwargs.pop("polling_interval", self._config.polling_interval) cont_token: Optional[str] = kwargs.pop("continuation_token", None) if cont_token is None: raw_result = self._create_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, ) 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: PollingMethod = cast( PollingMethod, LROBasePolling( lro_delay, path_format_arguments=path_format_arguments, **kwargs ), ) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.ContentAnalyzer].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.ContentAnalyzer]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) @distributed_trace def delete_analyzer( # pylint: disable=inconsistent-return-statements 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 = ( self._client._pipeline.run( # 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 def delete_result( self, operation_id: str, **kwargs: Any ) -> None: # pylint: disable=inconsistent-return-statements """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 = ( self._client._pipeline.run( # 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 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 = ( self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error( status_code=response.status_code, response=response, error_map=error_map ) 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 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 = ( self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error( status_code=response.status_code, response=response, error_map=error_map ) 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 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 = ( self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error( status_code=response.status_code, response=response, error_map=error_map ) 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 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 = ( self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error( status_code=response.status_code, response=response, error_map=error_map ) 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 def get_result_file( self, operation_id: str, path: str, **kwargs: Any ) -> Iterator[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: Iterator[bytes] :rtype: Iterator[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[Iterator[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 = ( self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error( status_code=response.status_code, response=response, error_map=error_map ) 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 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 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 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 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 = ( self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error( status_code=response.status_code, response=response, error_map=error_map ) 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) -> ItemPaged["_models.ContentAnalyzer"]: """List analyzers. :return: An iterator like instance of ContentAnalyzer :rtype: ~azure.core.paging.ItemPaged[~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 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, iter(list_of_elem) def get_next(next_link=None): _request = prepare_request(next_link) _stream = False pipeline_response: PipelineResponse = ( self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) ) response = pipeline_response.http_response if response.status_code not in [200]: map_error( status_code=response.status_code, response=response, error_map=error_map, ) raise HttpResponseError(response=response) return pipeline_response return ItemPaged(get_next, extract_data) @overload 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 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 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 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 = ( self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error( status_code=response.status_code, response=response, error_map=error_map ) 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 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 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 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 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 = ( self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error( status_code=response.status_code, response=response, error_map=error_map ) 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