# pylint: disable=too-many-lines # coding=utf-8 # -------------------------------------------------------------------------- # Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the MIT License. See License.txt in the project root for license information. # Code generated by Microsoft (R) Python Code Generator. # Changes may cause incorrect behavior and will be lost if the code is regenerated. # -------------------------------------------------------------------------- from io import IOBase import json import sys from typing import Any, Callable, Dict, IO, Iterable, Iterator, List, Optional, TypeVar, Union, cast, overload import urllib.parse 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 .._model_base import SdkJSONEncoder, _deserialize, _failsafe_deserialize from .._serialization import Serializer from .._vendor import DocumentIntelligenceAdministrationClientMixinABC, DocumentIntelligenceClientMixinABC if sys.version_info >= (3, 9): from collections.abc import MutableMapping else: from typing import MutableMapping # type: ignore JSON = MutableMapping[str, Any] # pylint: disable=unsubscriptable-object T = TypeVar("T") ClsType = Optional[Callable[[PipelineResponse[HttpRequest, HttpResponse], T, Dict[str, Any]], Any]] _SERIALIZER = Serializer() _SERIALIZER.client_side_validation = False def build_document_intelligence_analyze_document_request( # pylint: disable=name-too-long model_id: str, *, pages: Optional[str] = None, locale: Optional[str] = None, string_index_type: Optional[Union[str, _models.StringIndexType]] = None, features: Optional[List[Union[str, _models.DocumentAnalysisFeature]]] = None, query_fields: Optional[List[str]] = None, output_content_format: Optional[Union[str, _models.DocumentContentFormat]] = None, output: Optional[List[Union[str, _models.AnalyzeOutputOption]]] = 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", "2024-11-30")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/documentModels/{modelId}:analyze" path_format_arguments = { "modelId": _SERIALIZER.url("model_id", model_id, "str"), } _url: str = _url.format(**path_format_arguments) # type: ignore # Construct parameters _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str") if pages is not None: _params["pages"] = _SERIALIZER.query("pages", pages, "str") if locale is not None: _params["locale"] = _SERIALIZER.query("locale", locale, "str") if string_index_type is not None: _params["stringIndexType"] = _SERIALIZER.query("string_index_type", string_index_type, "str") if features is not None: _params["features"] = _SERIALIZER.query("features", features, "[str]", div=",") if query_fields is not None: _params["queryFields"] = _SERIALIZER.query("query_fields", query_fields, "[str]", div=",") if output_content_format is not None: _params["outputContentFormat"] = _SERIALIZER.query("output_content_format", output_content_format, "str") if output is not None: _params["output"] = _SERIALIZER.query("output", output, "[str]", div=",") # 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_document_intelligence_get_analyze_result_pdf_request( # pylint: disable=name-too-long model_id: str, result_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", "2024-11-30")) accept = _headers.pop("Accept", "application/pdf") # Construct URL _url = "/documentModels/{modelId}/analyzeResults/{resultId}/pdf" path_format_arguments = { "modelId": _SERIALIZER.url("model_id", model_id, "str"), "resultId": _SERIALIZER.url("result_id", result_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_document_intelligence_get_analyze_result_figure_request( # pylint: disable=name-too-long model_id: str, result_id: str, figure_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", "2024-11-30")) accept = _headers.pop("Accept", "image/png") # Construct URL _url = "/documentModels/{modelId}/analyzeResults/{resultId}/figures/{figureId}" path_format_arguments = { "modelId": _SERIALIZER.url("model_id", model_id, "str"), "resultId": _SERIALIZER.url("result_id", result_id, "str"), "figureId": _SERIALIZER.url("figure_id", figure_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_document_intelligence_delete_analyze_result_request( # pylint: disable=name-too-long model_id: str, result_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", "2024-11-30")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/documentModels/{modelId}/analyzeResults/{resultId}" path_format_arguments = { "modelId": _SERIALIZER.url("model_id", model_id, "str"), "resultId": _SERIALIZER.url("result_id", result_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="DELETE", url=_url, params=_params, headers=_headers, **kwargs) def build_document_intelligence_analyze_batch_documents_request( # pylint: disable=name-too-long model_id: str, *, pages: Optional[str] = None, locale: Optional[str] = None, string_index_type: Optional[Union[str, _models.StringIndexType]] = None, features: Optional[List[Union[str, _models.DocumentAnalysisFeature]]] = None, query_fields: Optional[List[str]] = None, output_content_format: Optional[Union[str, _models.DocumentContentFormat]] = None, output: Optional[List[Union[str, _models.AnalyzeOutputOption]]] = 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", "2024-11-30")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/documentModels/{modelId}:analyzeBatch" path_format_arguments = { "modelId": _SERIALIZER.url("model_id", model_id, "str"), } _url: str = _url.format(**path_format_arguments) # type: ignore # Construct parameters _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str") if pages is not None: _params["pages"] = _SERIALIZER.query("pages", pages, "str") if locale is not None: _params["locale"] = _SERIALIZER.query("locale", locale, "str") if string_index_type is not None: _params["stringIndexType"] = _SERIALIZER.query("string_index_type", string_index_type, "str") if features is not None: _params["features"] = _SERIALIZER.query("features", features, "[str]", div=",") if query_fields is not None: _params["queryFields"] = _SERIALIZER.query("query_fields", query_fields, "[str]", div=",") if output_content_format is not None: _params["outputContentFormat"] = _SERIALIZER.query("output_content_format", output_content_format, "str") if output is not None: _params["output"] = _SERIALIZER.query("output", output, "[str]", div=",") # 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_document_intelligence_list_analyze_batch_results_request( # pylint: disable=name-too-long model_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", "2024-11-30")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/documentModels/{modelId}/analyzeBatchResults" path_format_arguments = { "modelId": _SERIALIZER.url("model_id", model_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_document_intelligence_delete_analyze_batch_result_request( # pylint: disable=name-too-long model_id: str, result_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", "2024-11-30")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/documentModels/{modelId}/analyzeBatchResults/{resultId}" path_format_arguments = { "modelId": _SERIALIZER.url("model_id", model_id, "str"), "resultId": _SERIALIZER.url("result_id", result_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="DELETE", url=_url, params=_params, headers=_headers, **kwargs) def build_document_intelligence_get_analyze_batch_result_request( # pylint: disable=name-too-long model_id: str, result_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", "2024-11-30")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/documentModels/{modelId}/analyzeBatchResults/{resultId}" path_format_arguments = { "modelId": _SERIALIZER.url("model_id", model_id, "str"), "resultId": _SERIALIZER.url("result_id", result_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_document_intelligence_classify_document_request( # pylint: disable=name-too-long classifier_id: str, *, string_index_type: Optional[Union[str, _models.StringIndexType]] = None, split: Optional[Union[str, _models.SplitMode]] = None, pages: Optional[str] = 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", "2024-11-30")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/documentClassifiers/{classifierId}:analyze" path_format_arguments = { "classifierId": _SERIALIZER.url("classifier_id", classifier_id, "str"), } _url: str = _url.format(**path_format_arguments) # type: ignore # Construct parameters _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str") if string_index_type is not None: _params["stringIndexType"] = _SERIALIZER.query("string_index_type", string_index_type, "str") if split is not None: _params["split"] = _SERIALIZER.query("split", split, "str") if pages is not None: _params["pages"] = _SERIALIZER.query("pages", pages, "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_document_intelligence_administration_build_document_model_request( # pylint: disable=name-too-long **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", "2024-11-30")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/documentModels:build" # 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_document_intelligence_administration_compose_model_request( # pylint: disable=name-too-long **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", "2024-11-30")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/documentModels:compose" # 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_document_intelligence_administration_authorize_model_copy_request( # pylint: disable=name-too-long **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", "2024-11-30")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/documentModels:authorizeCopy" # 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_document_intelligence_administration_copy_model_to_request( # pylint: disable=name-too-long model_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", "2024-11-30")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/documentModels/{modelId}:copyTo" path_format_arguments = { "modelId": _SERIALIZER.url("model_id", model_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_document_intelligence_administration_get_model_request( # pylint: disable=name-too-long model_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", "2024-11-30")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/documentModels/{modelId}" path_format_arguments = { "modelId": _SERIALIZER.url("model_id", model_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_document_intelligence_administration_list_models_request( # pylint: disable=name-too-long **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", "2024-11-30")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/documentModels" # 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_document_intelligence_administration_delete_model_request( # pylint: disable=name-too-long model_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", "2024-11-30")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/documentModels/{modelId}" path_format_arguments = { "modelId": _SERIALIZER.url("model_id", model_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="DELETE", url=_url, params=_params, headers=_headers, **kwargs) def build_document_intelligence_administration_get_resource_details_request( # pylint: disable=name-too-long **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", "2024-11-30")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/info" # 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_document_intelligence_administration_get_operation_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", "2024-11-30")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/operations/{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_document_intelligence_administration_list_operations_request( # pylint: disable=name-too-long **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", "2024-11-30")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/operations" # 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_document_intelligence_administration_build_classifier_request( # pylint: disable=name-too-long **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", "2024-11-30")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/documentClassifiers:build" # 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_document_intelligence_administration_authorize_classifier_copy_request( # pylint: disable=name-too-long **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", "2024-11-30")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/documentClassifiers:authorizeCopy" # 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_document_intelligence_administration_copy_classifier_to_request( # pylint: disable=name-too-long classifier_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", "2024-11-30")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/documentClassifiers/{classifierId}:copyTo" path_format_arguments = { "classifierId": _SERIALIZER.url("classifier_id", classifier_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_document_intelligence_administration_get_classifier_request( # pylint: disable=name-too-long classifier_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", "2024-11-30")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/documentClassifiers/{classifierId}" path_format_arguments = { "classifierId": _SERIALIZER.url("classifier_id", classifier_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_document_intelligence_administration_list_classifiers_request( # pylint: disable=name-too-long **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", "2024-11-30")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/documentClassifiers" # 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_document_intelligence_administration_delete_classifier_request( # pylint: disable=name-too-long classifier_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", "2024-11-30")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/documentClassifiers/{classifierId}" path_format_arguments = { "classifierId": _SERIALIZER.url("classifier_id", classifier_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="DELETE", url=_url, params=_params, headers=_headers, **kwargs) class DocumentIntelligenceClientOperationsMixin(DocumentIntelligenceClientMixinABC): # pylint: disable=name-too-long def _analyze_document_initial( self, model_id: str, body: Union[_models.AnalyzeDocumentRequest, JSON, IO[bytes]], *, pages: Optional[str] = None, locale: Optional[str] = None, string_index_type: Optional[Union[str, _models.StringIndexType]] = None, features: Optional[List[Union[str, _models.DocumentAnalysisFeature]]] = None, query_fields: Optional[List[str]] = None, output_content_format: Optional[Union[str, _models.DocumentContentFormat]] = None, output: Optional[List[Union[str, _models.AnalyzeOutputOption]]] = 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(body, (IOBase, bytes)): _content = body else: _content = json.dumps(body, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore _request = build_document_intelligence_analyze_document_request( model_id=model_id, pages=pages, locale=locale, string_index_type=string_index_type, features=features, query_fields=query_fields, output_content_format=output_content_format, output=output, 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) _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) error = _failsafe_deserialize(_models.DocumentIntelligenceErrorResponse, response.json()) raise HttpResponseError(response=response, model=error) response_headers = {} response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) response_headers["Operation-Location"] = self._deserialize("str", response.headers.get("Operation-Location")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @overload def begin_analyze_document( self, model_id: str, body: _models.AnalyzeDocumentRequest, *, pages: Optional[str] = None, locale: Optional[str] = None, string_index_type: Optional[Union[str, _models.StringIndexType]] = None, features: Optional[List[Union[str, _models.DocumentAnalysisFeature]]] = None, query_fields: Optional[List[str]] = None, output_content_format: Optional[Union[str, _models.DocumentContentFormat]] = None, output: Optional[List[Union[str, _models.AnalyzeOutputOption]]] = None, content_type: str = "application/json", **kwargs: Any, ) -> LROPoller[_models.AnalyzeResult]: """Analyzes document with document model. :param model_id: Unique document model name. Required. :type model_id: str :param body: Analyze request parameters. Required. :type body: ~azure.ai.documentintelligence.models.AnalyzeDocumentRequest :keyword pages: 1-based page numbers to analyze. Ex. "1-3,5,7-9". Default value is None. :paramtype pages: str :keyword locale: Locale hint for text recognition and document analysis. Value may contain only the language code (ex. "en", "fr") or BCP 47 language tag (ex. "en-US"). Default value is None. :paramtype locale: str :keyword string_index_type: Method used to compute string offset and length. Known values are: "textElements", "unicodeCodePoint", and "utf16CodeUnit". Default value is None. :paramtype string_index_type: str or ~azure.ai.documentintelligence.models.StringIndexType :keyword features: List of optional analysis features. Default value is None. :paramtype features: list[str or ~azure.ai.documentintelligence.models.DocumentAnalysisFeature] :keyword query_fields: List of additional fields to extract. Ex. "NumberOfGuests,StoreNumber". Default value is None. :paramtype query_fields: list[str] :keyword output_content_format: Format of the analyze result top-level content. Known values are: "text" and "markdown". Default value is None. :paramtype output_content_format: str or ~azure.ai.documentintelligence.models.DocumentContentFormat :keyword output: Additional outputs to generate during analysis. Default value is None. :paramtype output: list[str or ~azure.ai.documentintelligence.models.AnalyzeOutputOption] :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 AnalyzeResult. The AnalyzeResult is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.AnalyzeResult] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_analyze_document( self, model_id: str, body: JSON, *, pages: Optional[str] = None, locale: Optional[str] = None, string_index_type: Optional[Union[str, _models.StringIndexType]] = None, features: Optional[List[Union[str, _models.DocumentAnalysisFeature]]] = None, query_fields: Optional[List[str]] = None, output_content_format: Optional[Union[str, _models.DocumentContentFormat]] = None, output: Optional[List[Union[str, _models.AnalyzeOutputOption]]] = None, content_type: str = "application/json", **kwargs: Any, ) -> LROPoller[_models.AnalyzeResult]: """Analyzes document with document model. :param model_id: Unique document model name. Required. :type model_id: str :param body: Analyze request parameters. Required. :type body: JSON :keyword pages: 1-based page numbers to analyze. Ex. "1-3,5,7-9". Default value is None. :paramtype pages: str :keyword locale: Locale hint for text recognition and document analysis. Value may contain only the language code (ex. "en", "fr") or BCP 47 language tag (ex. "en-US"). Default value is None. :paramtype locale: str :keyword string_index_type: Method used to compute string offset and length. Known values are: "textElements", "unicodeCodePoint", and "utf16CodeUnit". Default value is None. :paramtype string_index_type: str or ~azure.ai.documentintelligence.models.StringIndexType :keyword features: List of optional analysis features. Default value is None. :paramtype features: list[str or ~azure.ai.documentintelligence.models.DocumentAnalysisFeature] :keyword query_fields: List of additional fields to extract. Ex. "NumberOfGuests,StoreNumber". Default value is None. :paramtype query_fields: list[str] :keyword output_content_format: Format of the analyze result top-level content. Known values are: "text" and "markdown". Default value is None. :paramtype output_content_format: str or ~azure.ai.documentintelligence.models.DocumentContentFormat :keyword output: Additional outputs to generate during analysis. Default value is None. :paramtype output: list[str or ~azure.ai.documentintelligence.models.AnalyzeOutputOption] :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 AnalyzeResult. The AnalyzeResult is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.AnalyzeResult] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_analyze_document( self, model_id: str, body: IO[bytes], *, pages: Optional[str] = None, locale: Optional[str] = None, string_index_type: Optional[Union[str, _models.StringIndexType]] = None, features: Optional[List[Union[str, _models.DocumentAnalysisFeature]]] = None, query_fields: Optional[List[str]] = None, output_content_format: Optional[Union[str, _models.DocumentContentFormat]] = None, output: Optional[List[Union[str, _models.AnalyzeOutputOption]]] = None, content_type: str = "application/json", **kwargs: Any, ) -> LROPoller[_models.AnalyzeResult]: """Analyzes document with document model. :param model_id: Unique document model name. Required. :type model_id: str :param body: Analyze request parameters. Required. :type body: IO[bytes] :keyword pages: 1-based page numbers to analyze. Ex. "1-3,5,7-9". Default value is None. :paramtype pages: str :keyword locale: Locale hint for text recognition and document analysis. Value may contain only the language code (ex. "en", "fr") or BCP 47 language tag (ex. "en-US"). Default value is None. :paramtype locale: str :keyword string_index_type: Method used to compute string offset and length. Known values are: "textElements", "unicodeCodePoint", and "utf16CodeUnit". Default value is None. :paramtype string_index_type: str or ~azure.ai.documentintelligence.models.StringIndexType :keyword features: List of optional analysis features. Default value is None. :paramtype features: list[str or ~azure.ai.documentintelligence.models.DocumentAnalysisFeature] :keyword query_fields: List of additional fields to extract. Ex. "NumberOfGuests,StoreNumber". Default value is None. :paramtype query_fields: list[str] :keyword output_content_format: Format of the analyze result top-level content. Known values are: "text" and "markdown". Default value is None. :paramtype output_content_format: str or ~azure.ai.documentintelligence.models.DocumentContentFormat :keyword output: Additional outputs to generate during analysis. Default value is None. :paramtype output: list[str or ~azure.ai.documentintelligence.models.AnalyzeOutputOption] :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 AnalyzeResult. The AnalyzeResult is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.AnalyzeResult] :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace def begin_analyze_document( self, model_id: str, body: Union[_models.AnalyzeDocumentRequest, JSON, IO[bytes]], *, pages: Optional[str] = None, locale: Optional[str] = None, string_index_type: Optional[Union[str, _models.StringIndexType]] = None, features: Optional[List[Union[str, _models.DocumentAnalysisFeature]]] = None, query_fields: Optional[List[str]] = None, output_content_format: Optional[Union[str, _models.DocumentContentFormat]] = None, output: Optional[List[Union[str, _models.AnalyzeOutputOption]]] = None, **kwargs: Any, ) -> LROPoller[_models.AnalyzeResult]: """Analyzes document with document model. :param model_id: Unique document model name. Required. :type model_id: str :param body: Analyze request parameters. Is one of the following types: AnalyzeDocumentRequest, JSON, IO[bytes] Required. :type body: ~azure.ai.documentintelligence.models.AnalyzeDocumentRequest or JSON or IO[bytes] :keyword pages: 1-based page numbers to analyze. Ex. "1-3,5,7-9". Default value is None. :paramtype pages: str :keyword locale: Locale hint for text recognition and document analysis. Value may contain only the language code (ex. "en", "fr") or BCP 47 language tag (ex. "en-US"). Default value is None. :paramtype locale: str :keyword string_index_type: Method used to compute string offset and length. Known values are: "textElements", "unicodeCodePoint", and "utf16CodeUnit". Default value is None. :paramtype string_index_type: str or ~azure.ai.documentintelligence.models.StringIndexType :keyword features: List of optional analysis features. Default value is None. :paramtype features: list[str or ~azure.ai.documentintelligence.models.DocumentAnalysisFeature] :keyword query_fields: List of additional fields to extract. Ex. "NumberOfGuests,StoreNumber". Default value is None. :paramtype query_fields: list[str] :keyword output_content_format: Format of the analyze result top-level content. Known values are: "text" and "markdown". Default value is None. :paramtype output_content_format: str or ~azure.ai.documentintelligence.models.DocumentContentFormat :keyword output: Additional outputs to generate during analysis. Default value is None. :paramtype output: list[str or ~azure.ai.documentintelligence.models.AnalyzeOutputOption] :return: An instance of LROPoller that returns AnalyzeResult. The AnalyzeResult is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.AnalyzeResult] :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.AnalyzeResult] = 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_document_initial( model_id=model_id, body=body, pages=pages, locale=locale, string_index_type=string_index_type, features=features, query_fields=query_fields, output_content_format=output_content_format, output=output, 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["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) response_headers["Operation-Location"] = self._deserialize( "str", response.headers.get("Operation-Location") ) deserialized = _deserialize(_models.AnalyzeResult, response.json().get("analyzeResult")) 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.AnalyzeResult].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.AnalyzeResult]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) @distributed_trace def get_analyze_result_pdf(self, model_id: str, result_id: str, **kwargs: Any) -> Iterator[bytes]: """Gets the generated searchable PDF output from document analysis. :param model_id: Unique document model name. Required. :type model_id: str :param result_id: Analyze operation result ID. Required. :type result_id: 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_document_intelligence_get_analyze_result_pdf_request( model_id=model_id, result_id=result_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 = 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) error = _failsafe_deserialize(_models.DocumentIntelligenceErrorResponse, response.json()) raise HttpResponseError(response=response, model=error) response_headers = {} response_headers["content-type"] = self._deserialize("str", response.headers.get("content-type")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @distributed_trace def get_analyze_result_figure( self, model_id: str, result_id: str, figure_id: str, **kwargs: Any ) -> Iterator[bytes]: """Gets the generated cropped image of specified figure from document analysis. :param model_id: Unique document model name. Required. :type model_id: str :param result_id: Analyze operation result ID. Required. :type result_id: str :param figure_id: Figure ID. Required. :type figure_id: 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_document_intelligence_get_analyze_result_figure_request( model_id=model_id, result_id=result_id, figure_id=figure_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 = 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) error = _failsafe_deserialize(_models.DocumentIntelligenceErrorResponse, response.json()) raise HttpResponseError(response=response, model=error) response_headers = {} response_headers["content-type"] = self._deserialize("str", response.headers.get("content-type")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @distributed_trace def delete_analyze_result( # pylint: disable=inconsistent-return-statements self, model_id: str, result_id: str, **kwargs: Any ) -> None: """Mark the result of document analysis for deletion. :param model_id: Unique document model name. Required. :type model_id: str :param result_id: Analyze operation result ID. Required. :type result_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_document_intelligence_delete_analyze_result_request( model_id=model_id, result_id=result_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) error = _failsafe_deserialize(_models.DocumentIntelligenceErrorResponse, response.json()) raise HttpResponseError(response=response, model=error) if cls: return cls(pipeline_response, None, {}) # type: ignore def _analyze_batch_documents_initial( self, model_id: str, body: Union[_models.AnalyzeBatchDocumentsRequest, JSON, IO[bytes]], *, pages: Optional[str] = None, locale: Optional[str] = None, string_index_type: Optional[Union[str, _models.StringIndexType]] = None, features: Optional[List[Union[str, _models.DocumentAnalysisFeature]]] = None, query_fields: Optional[List[str]] = None, output_content_format: Optional[Union[str, _models.DocumentContentFormat]] = None, output: Optional[List[Union[str, _models.AnalyzeOutputOption]]] = 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(body, (IOBase, bytes)): _content = body else: _content = json.dumps(body, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore _request = build_document_intelligence_analyze_batch_documents_request( model_id=model_id, pages=pages, locale=locale, string_index_type=string_index_type, features=features, query_fields=query_fields, output_content_format=output_content_format, output=output, 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) _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) error = _failsafe_deserialize(_models.DocumentIntelligenceErrorResponse, response.json()) raise HttpResponseError(response=response, model=error) response_headers = {} response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) response_headers["Operation-Location"] = self._deserialize("str", response.headers.get("Operation-Location")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @overload def begin_analyze_batch_documents( self, model_id: str, body: _models.AnalyzeBatchDocumentsRequest, *, pages: Optional[str] = None, locale: Optional[str] = None, string_index_type: Optional[Union[str, _models.StringIndexType]] = None, features: Optional[List[Union[str, _models.DocumentAnalysisFeature]]] = None, query_fields: Optional[List[str]] = None, output_content_format: Optional[Union[str, _models.DocumentContentFormat]] = None, output: Optional[List[Union[str, _models.AnalyzeOutputOption]]] = None, content_type: str = "application/json", **kwargs: Any, ) -> LROPoller[_models.AnalyzeBatchResult]: """Analyzes batch documents with document model. :param model_id: Unique document model name. Required. :type model_id: str :param body: Analyze batch request parameters. Required. :type body: ~azure.ai.documentintelligence.models.AnalyzeBatchDocumentsRequest :keyword pages: 1-based page numbers to analyze. Ex. "1-3,5,7-9". Default value is None. :paramtype pages: str :keyword locale: Locale hint for text recognition and document analysis. Value may contain only the language code (ex. "en", "fr") or BCP 47 language tag (ex. "en-US"). Default value is None. :paramtype locale: str :keyword string_index_type: Method used to compute string offset and length. Known values are: "textElements", "unicodeCodePoint", and "utf16CodeUnit". Default value is None. :paramtype string_index_type: str or ~azure.ai.documentintelligence.models.StringIndexType :keyword features: List of optional analysis features. Default value is None. :paramtype features: list[str or ~azure.ai.documentintelligence.models.DocumentAnalysisFeature] :keyword query_fields: List of additional fields to extract. Ex. "NumberOfGuests,StoreNumber". Default value is None. :paramtype query_fields: list[str] :keyword output_content_format: Format of the analyze result top-level content. Known values are: "text" and "markdown". Default value is None. :paramtype output_content_format: str or ~azure.ai.documentintelligence.models.DocumentContentFormat :keyword output: Additional outputs to generate during analysis. Default value is None. :paramtype output: list[str or ~azure.ai.documentintelligence.models.AnalyzeOutputOption] :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 AnalyzeBatchResult. The AnalyzeBatchResult is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.AnalyzeBatchResult] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_analyze_batch_documents( self, model_id: str, body: JSON, *, pages: Optional[str] = None, locale: Optional[str] = None, string_index_type: Optional[Union[str, _models.StringIndexType]] = None, features: Optional[List[Union[str, _models.DocumentAnalysisFeature]]] = None, query_fields: Optional[List[str]] = None, output_content_format: Optional[Union[str, _models.DocumentContentFormat]] = None, output: Optional[List[Union[str, _models.AnalyzeOutputOption]]] = None, content_type: str = "application/json", **kwargs: Any, ) -> LROPoller[_models.AnalyzeBatchResult]: """Analyzes batch documents with document model. :param model_id: Unique document model name. Required. :type model_id: str :param body: Analyze batch request parameters. Required. :type body: JSON :keyword pages: 1-based page numbers to analyze. Ex. "1-3,5,7-9". Default value is None. :paramtype pages: str :keyword locale: Locale hint for text recognition and document analysis. Value may contain only the language code (ex. "en", "fr") or BCP 47 language tag (ex. "en-US"). Default value is None. :paramtype locale: str :keyword string_index_type: Method used to compute string offset and length. Known values are: "textElements", "unicodeCodePoint", and "utf16CodeUnit". Default value is None. :paramtype string_index_type: str or ~azure.ai.documentintelligence.models.StringIndexType :keyword features: List of optional analysis features. Default value is None. :paramtype features: list[str or ~azure.ai.documentintelligence.models.DocumentAnalysisFeature] :keyword query_fields: List of additional fields to extract. Ex. "NumberOfGuests,StoreNumber". Default value is None. :paramtype query_fields: list[str] :keyword output_content_format: Format of the analyze result top-level content. Known values are: "text" and "markdown". Default value is None. :paramtype output_content_format: str or ~azure.ai.documentintelligence.models.DocumentContentFormat :keyword output: Additional outputs to generate during analysis. Default value is None. :paramtype output: list[str or ~azure.ai.documentintelligence.models.AnalyzeOutputOption] :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 AnalyzeBatchResult. The AnalyzeBatchResult is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.AnalyzeBatchResult] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_analyze_batch_documents( self, model_id: str, body: IO[bytes], *, pages: Optional[str] = None, locale: Optional[str] = None, string_index_type: Optional[Union[str, _models.StringIndexType]] = None, features: Optional[List[Union[str, _models.DocumentAnalysisFeature]]] = None, query_fields: Optional[List[str]] = None, output_content_format: Optional[Union[str, _models.DocumentContentFormat]] = None, output: Optional[List[Union[str, _models.AnalyzeOutputOption]]] = None, content_type: str = "application/json", **kwargs: Any, ) -> LROPoller[_models.AnalyzeBatchResult]: """Analyzes batch documents with document model. :param model_id: Unique document model name. Required. :type model_id: str :param body: Analyze batch request parameters. Required. :type body: IO[bytes] :keyword pages: 1-based page numbers to analyze. Ex. "1-3,5,7-9". Default value is None. :paramtype pages: str :keyword locale: Locale hint for text recognition and document analysis. Value may contain only the language code (ex. "en", "fr") or BCP 47 language tag (ex. "en-US"). Default value is None. :paramtype locale: str :keyword string_index_type: Method used to compute string offset and length. Known values are: "textElements", "unicodeCodePoint", and "utf16CodeUnit". Default value is None. :paramtype string_index_type: str or ~azure.ai.documentintelligence.models.StringIndexType :keyword features: List of optional analysis features. Default value is None. :paramtype features: list[str or ~azure.ai.documentintelligence.models.DocumentAnalysisFeature] :keyword query_fields: List of additional fields to extract. Ex. "NumberOfGuests,StoreNumber". Default value is None. :paramtype query_fields: list[str] :keyword output_content_format: Format of the analyze result top-level content. Known values are: "text" and "markdown". Default value is None. :paramtype output_content_format: str or ~azure.ai.documentintelligence.models.DocumentContentFormat :keyword output: Additional outputs to generate during analysis. Default value is None. :paramtype output: list[str or ~azure.ai.documentintelligence.models.AnalyzeOutputOption] :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 AnalyzeBatchResult. The AnalyzeBatchResult is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.AnalyzeBatchResult] :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace def begin_analyze_batch_documents( self, model_id: str, body: Union[_models.AnalyzeBatchDocumentsRequest, JSON, IO[bytes]], *, pages: Optional[str] = None, locale: Optional[str] = None, string_index_type: Optional[Union[str, _models.StringIndexType]] = None, features: Optional[List[Union[str, _models.DocumentAnalysisFeature]]] = None, query_fields: Optional[List[str]] = None, output_content_format: Optional[Union[str, _models.DocumentContentFormat]] = None, output: Optional[List[Union[str, _models.AnalyzeOutputOption]]] = None, **kwargs: Any, ) -> LROPoller[_models.AnalyzeBatchResult]: """Analyzes batch documents with document model. :param model_id: Unique document model name. Required. :type model_id: str :param body: Analyze batch request parameters. Is one of the following types: AnalyzeBatchDocumentsRequest, JSON, IO[bytes] Required. :type body: ~azure.ai.documentintelligence.models.AnalyzeBatchDocumentsRequest or JSON or IO[bytes] :keyword pages: 1-based page numbers to analyze. Ex. "1-3,5,7-9". Default value is None. :paramtype pages: str :keyword locale: Locale hint for text recognition and document analysis. Value may contain only the language code (ex. "en", "fr") or BCP 47 language tag (ex. "en-US"). Default value is None. :paramtype locale: str :keyword string_index_type: Method used to compute string offset and length. Known values are: "textElements", "unicodeCodePoint", and "utf16CodeUnit". Default value is None. :paramtype string_index_type: str or ~azure.ai.documentintelligence.models.StringIndexType :keyword features: List of optional analysis features. Default value is None. :paramtype features: list[str or ~azure.ai.documentintelligence.models.DocumentAnalysisFeature] :keyword query_fields: List of additional fields to extract. Ex. "NumberOfGuests,StoreNumber". Default value is None. :paramtype query_fields: list[str] :keyword output_content_format: Format of the analyze result top-level content. Known values are: "text" and "markdown". Default value is None. :paramtype output_content_format: str or ~azure.ai.documentintelligence.models.DocumentContentFormat :keyword output: Additional outputs to generate during analysis. Default value is None. :paramtype output: list[str or ~azure.ai.documentintelligence.models.AnalyzeOutputOption] :return: An instance of LROPoller that returns AnalyzeBatchResult. The AnalyzeBatchResult is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.AnalyzeBatchResult] :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.AnalyzeBatchResult] = 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_batch_documents_initial( model_id=model_id, body=body, pages=pages, locale=locale, string_index_type=string_index_type, features=features, query_fields=query_fields, output_content_format=output_content_format, output=output, 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["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) response_headers["Operation-Location"] = self._deserialize( "str", response.headers.get("Operation-Location") ) deserialized = _deserialize(_models.AnalyzeBatchResult, 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.AnalyzeBatchResult].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.AnalyzeBatchResult]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) @distributed_trace def list_analyze_batch_results(self, model_id: str, **kwargs: Any) -> Iterable["_models.AnalyzeBatchOperation"]: """List batch document analysis results. :param model_id: Unique document model name. Required. :type model_id: str :return: An iterator like instance of AnalyzeBatchOperation :rtype: ~azure.core.paging.ItemPaged[~azure.ai.documentintelligence.models.AnalyzeBatchOperation] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[List[_models.AnalyzeBatchOperation]] = 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_document_intelligence_list_analyze_batch_results_request( model_id=model_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) 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.AnalyzeBatchOperation], deserialized["value"]) if cls: list_of_elem = cls(list_of_elem) # type: ignore return deserialized.get("nextLink") or None, iter(list_of_elem) def get_next(next_link=None): _request = prepare_request(next_link) _stream = False pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.DocumentIntelligenceErrorResponse, response.json()) raise HttpResponseError(response=response, model=error) return pipeline_response return ItemPaged(get_next, extract_data) @distributed_trace def delete_analyze_batch_result( # pylint: disable=inconsistent-return-statements self, model_id: str, result_id: str, **kwargs: Any ) -> None: """Mark the batch document analysis result for deletion. :param model_id: Unique document model name. Required. :type model_id: str :param result_id: Analyze batch operation result ID. Required. :type result_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_document_intelligence_delete_analyze_batch_result_request( model_id=model_id, result_id=result_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) error = _failsafe_deserialize(_models.DocumentIntelligenceErrorResponse, response.json()) raise HttpResponseError(response=response, model=error) if cls: return cls(pipeline_response, None, {}) # type: ignore @distributed_trace def get_analyze_batch_result(self, model_id: str, result_id: str, **kwargs: Any) -> _models.AnalyzeBatchOperation: """Gets the result of batch document analysis. :param model_id: Unique document model name. Required. :type model_id: str :param result_id: Analyze batch operation result ID. Required. :type result_id: str :return: AnalyzeBatchOperation. The AnalyzeBatchOperation is compatible with MutableMapping :rtype: ~azure.ai.documentintelligence.models.AnalyzeBatchOperation :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.AnalyzeBatchOperation] = kwargs.pop("cls", None) _request = build_document_intelligence_get_analyze_batch_result_request( model_id=model_id, result_id=result_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 = kwargs.pop("stream", False) pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.DocumentIntelligenceErrorResponse, response.json()) raise HttpResponseError(response=response, model=error) if _stream: deserialized = response.iter_bytes() else: deserialized = _deserialize(_models.AnalyzeBatchOperation, response.json()) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore def _classify_document_initial( self, classifier_id: str, body: Union[_models.ClassifyDocumentRequest, JSON, IO[bytes]], *, string_index_type: Optional[Union[str, _models.StringIndexType]] = None, split: Optional[Union[str, _models.SplitMode]] = None, pages: 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) 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_document_intelligence_classify_document_request( classifier_id=classifier_id, string_index_type=string_index_type, split=split, pages=pages, 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) _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) error = _failsafe_deserialize(_models.DocumentIntelligenceErrorResponse, response.json()) raise HttpResponseError(response=response, model=error) response_headers = {} response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) response_headers["Operation-Location"] = self._deserialize("str", response.headers.get("Operation-Location")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @overload def begin_classify_document( self, classifier_id: str, body: _models.ClassifyDocumentRequest, *, string_index_type: Optional[Union[str, _models.StringIndexType]] = None, split: Optional[Union[str, _models.SplitMode]] = None, pages: Optional[str] = None, content_type: str = "application/json", **kwargs: Any, ) -> LROPoller[_models.AnalyzeResult]: """Classifies document with document classifier. :param classifier_id: Unique document classifier name. Required. :type classifier_id: str :param body: Classify request parameters. Required. :type body: ~azure.ai.documentintelligence.models.ClassifyDocumentRequest :keyword string_index_type: Method used to compute string offset and length. Known values are: "textElements", "unicodeCodePoint", and "utf16CodeUnit". Default value is None. :paramtype string_index_type: str or ~azure.ai.documentintelligence.models.StringIndexType :keyword split: Document splitting mode. Known values are: "auto", "none", and "perPage". Default value is None. :paramtype split: str or ~azure.ai.documentintelligence.models.SplitMode :keyword pages: 1-based page numbers to analyze. Ex. "1-3,5,7-9". Default value is None. :paramtype pages: str :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 AnalyzeResult. The AnalyzeResult is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.AnalyzeResult] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_classify_document( self, classifier_id: str, body: JSON, *, string_index_type: Optional[Union[str, _models.StringIndexType]] = None, split: Optional[Union[str, _models.SplitMode]] = None, pages: Optional[str] = None, content_type: str = "application/json", **kwargs: Any, ) -> LROPoller[_models.AnalyzeResult]: """Classifies document with document classifier. :param classifier_id: Unique document classifier name. Required. :type classifier_id: str :param body: Classify request parameters. Required. :type body: JSON :keyword string_index_type: Method used to compute string offset and length. Known values are: "textElements", "unicodeCodePoint", and "utf16CodeUnit". Default value is None. :paramtype string_index_type: str or ~azure.ai.documentintelligence.models.StringIndexType :keyword split: Document splitting mode. Known values are: "auto", "none", and "perPage". Default value is None. :paramtype split: str or ~azure.ai.documentintelligence.models.SplitMode :keyword pages: 1-based page numbers to analyze. Ex. "1-3,5,7-9". Default value is None. :paramtype pages: str :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 AnalyzeResult. The AnalyzeResult is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.AnalyzeResult] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_classify_document( self, classifier_id: str, body: IO[bytes], *, string_index_type: Optional[Union[str, _models.StringIndexType]] = None, split: Optional[Union[str, _models.SplitMode]] = None, pages: Optional[str] = None, content_type: str = "application/json", **kwargs: Any, ) -> LROPoller[_models.AnalyzeResult]: """Classifies document with document classifier. :param classifier_id: Unique document classifier name. Required. :type classifier_id: str :param body: Classify request parameters. Required. :type body: IO[bytes] :keyword string_index_type: Method used to compute string offset and length. Known values are: "textElements", "unicodeCodePoint", and "utf16CodeUnit". Default value is None. :paramtype string_index_type: str or ~azure.ai.documentintelligence.models.StringIndexType :keyword split: Document splitting mode. Known values are: "auto", "none", and "perPage". Default value is None. :paramtype split: str or ~azure.ai.documentintelligence.models.SplitMode :keyword pages: 1-based page numbers to analyze. Ex. "1-3,5,7-9". Default value is None. :paramtype pages: str :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 AnalyzeResult. The AnalyzeResult is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.AnalyzeResult] :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace def begin_classify_document( self, classifier_id: str, body: Union[_models.ClassifyDocumentRequest, JSON, IO[bytes]], *, string_index_type: Optional[Union[str, _models.StringIndexType]] = None, split: Optional[Union[str, _models.SplitMode]] = None, pages: Optional[str] = None, **kwargs: Any, ) -> LROPoller[_models.AnalyzeResult]: """Classifies document with document classifier. :param classifier_id: Unique document classifier name. Required. :type classifier_id: str :param body: Classify request parameters. Is one of the following types: ClassifyDocumentRequest, JSON, IO[bytes] Required. :type body: ~azure.ai.documentintelligence.models.ClassifyDocumentRequest or JSON or IO[bytes] :keyword string_index_type: Method used to compute string offset and length. Known values are: "textElements", "unicodeCodePoint", and "utf16CodeUnit". Default value is None. :paramtype string_index_type: str or ~azure.ai.documentintelligence.models.StringIndexType :keyword split: Document splitting mode. Known values are: "auto", "none", and "perPage". Default value is None. :paramtype split: str or ~azure.ai.documentintelligence.models.SplitMode :keyword pages: 1-based page numbers to analyze. Ex. "1-3,5,7-9". Default value is None. :paramtype pages: str :return: An instance of LROPoller that returns AnalyzeResult. The AnalyzeResult is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.AnalyzeResult] :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.AnalyzeResult] = 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._classify_document_initial( classifier_id=classifier_id, body=body, string_index_type=string_index_type, split=split, pages=pages, 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["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) response_headers["Operation-Location"] = self._deserialize( "str", response.headers.get("Operation-Location") ) deserialized = _deserialize(_models.AnalyzeResult, response.json().get("analyzeResult")) 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.AnalyzeResult].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.AnalyzeResult]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) class DocumentIntelligenceAdministrationClientOperationsMixin( # pylint: disable=too-many-public-methods,name-too-long DocumentIntelligenceAdministrationClientMixinABC ): def _build_document_model_initial( self, body: Union[_models.BuildDocumentModelRequest, JSON, IO[bytes]], **kwargs: Any ) -> Iterator[bytes]: error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } error_map.update(kwargs.pop("error_map", {}) or {}) _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = kwargs.pop("params", {}) or {} content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) content_type = content_type or "application/json" _content = None if isinstance(body, (IOBase, bytes)): _content = body else: _content = json.dumps(body, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore _request = build_document_intelligence_administration_build_document_model_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) _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) error = _failsafe_deserialize(_models.DocumentIntelligenceErrorResponse, response.json()) raise HttpResponseError(response=response, model=error) response_headers = {} response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) response_headers["Operation-Location"] = self._deserialize("str", response.headers.get("Operation-Location")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @overload def begin_build_document_model( self, body: _models.BuildDocumentModelRequest, *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.DocumentModelDetails]: """Builds a custom document analysis model. :param body: Build request parameters. Required. :type body: ~azure.ai.documentintelligence.models.BuildDocumentModelRequest :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 DocumentModelDetails. The DocumentModelDetails is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.DocumentModelDetails] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_build_document_model( self, body: JSON, *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.DocumentModelDetails]: """Builds a custom document analysis model. :param body: Build request parameters. 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: An instance of LROPoller that returns DocumentModelDetails. The DocumentModelDetails is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.DocumentModelDetails] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_build_document_model( self, body: IO[bytes], *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.DocumentModelDetails]: """Builds a custom document analysis model. :param body: Build request parameters. 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: An instance of LROPoller that returns DocumentModelDetails. The DocumentModelDetails is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.DocumentModelDetails] :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace def begin_build_document_model( self, body: Union[_models.BuildDocumentModelRequest, JSON, IO[bytes]], **kwargs: Any ) -> LROPoller[_models.DocumentModelDetails]: """Builds a custom document analysis model. :param body: Build request parameters. Is one of the following types: BuildDocumentModelRequest, JSON, IO[bytes] Required. :type body: ~azure.ai.documentintelligence.models.BuildDocumentModelRequest or JSON or IO[bytes] :return: An instance of LROPoller that returns DocumentModelDetails. The DocumentModelDetails is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.DocumentModelDetails] :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.DocumentModelDetails] = 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._build_document_model_initial( body=body, 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["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) response_headers["Operation-Location"] = self._deserialize( "str", response.headers.get("Operation-Location") ) deserialized = _deserialize(_models.DocumentModelDetails, 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.DocumentModelDetails].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.DocumentModelDetails]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) def _compose_model_initial( self, body: Union[_models.ComposeDocumentModelRequest, JSON, IO[bytes]], **kwargs: Any ) -> Iterator[bytes]: error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } error_map.update(kwargs.pop("error_map", {}) or {}) _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = kwargs.pop("params", {}) or {} content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) content_type = content_type or "application/json" _content = None if isinstance(body, (IOBase, bytes)): _content = body else: _content = json.dumps(body, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore _request = build_document_intelligence_administration_compose_model_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) _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) error = _failsafe_deserialize(_models.DocumentIntelligenceErrorResponse, response.json()) raise HttpResponseError(response=response, model=error) response_headers = {} response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) response_headers["Operation-Location"] = self._deserialize("str", response.headers.get("Operation-Location")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @overload def begin_compose_model( self, body: _models.ComposeDocumentModelRequest, *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.DocumentModelDetails]: """Creates a new document model from document types of existing document models. :param body: Compose request parameters. Required. :type body: ~azure.ai.documentintelligence.models.ComposeDocumentModelRequest :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 DocumentModelDetails. The DocumentModelDetails is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.DocumentModelDetails] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_compose_model( self, body: JSON, *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.DocumentModelDetails]: """Creates a new document model from document types of existing document models. :param body: Compose request parameters. 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: An instance of LROPoller that returns DocumentModelDetails. The DocumentModelDetails is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.DocumentModelDetails] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_compose_model( self, body: IO[bytes], *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.DocumentModelDetails]: """Creates a new document model from document types of existing document models. :param body: Compose request parameters. 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: An instance of LROPoller that returns DocumentModelDetails. The DocumentModelDetails is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.DocumentModelDetails] :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace def begin_compose_model( self, body: Union[_models.ComposeDocumentModelRequest, JSON, IO[bytes]], **kwargs: Any ) -> LROPoller[_models.DocumentModelDetails]: """Creates a new document model from document types of existing document models. :param body: Compose request parameters. Is one of the following types: ComposeDocumentModelRequest, JSON, IO[bytes] Required. :type body: ~azure.ai.documentintelligence.models.ComposeDocumentModelRequest or JSON or IO[bytes] :return: An instance of LROPoller that returns DocumentModelDetails. The DocumentModelDetails is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.DocumentModelDetails] :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.DocumentModelDetails] = 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._compose_model_initial( body=body, 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["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) response_headers["Operation-Location"] = self._deserialize( "str", response.headers.get("Operation-Location") ) deserialized = _deserialize(_models.DocumentModelDetails, 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.DocumentModelDetails].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.DocumentModelDetails]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) @overload def authorize_model_copy( self, body: _models.AuthorizeCopyRequest, *, content_type: str = "application/json", **kwargs: Any ) -> _models.ModelCopyAuthorization: """Generates authorization to copy a document model to this location with specified modelId and optional description. :param body: Authorize copy request parameters. Required. :type body: ~azure.ai.documentintelligence.models.AuthorizeCopyRequest :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :return: ModelCopyAuthorization. The ModelCopyAuthorization is compatible with MutableMapping :rtype: ~azure.ai.documentintelligence.models.ModelCopyAuthorization :raises ~azure.core.exceptions.HttpResponseError: """ @overload def authorize_model_copy( self, body: JSON, *, content_type: str = "application/json", **kwargs: Any ) -> _models.ModelCopyAuthorization: """Generates authorization to copy a document model to this location with specified modelId and optional description. :param body: Authorize copy request parameters. 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: ModelCopyAuthorization. The ModelCopyAuthorization is compatible with MutableMapping :rtype: ~azure.ai.documentintelligence.models.ModelCopyAuthorization :raises ~azure.core.exceptions.HttpResponseError: """ @overload def authorize_model_copy( self, body: IO[bytes], *, content_type: str = "application/json", **kwargs: Any ) -> _models.ModelCopyAuthorization: """Generates authorization to copy a document model to this location with specified modelId and optional description. :param body: Authorize copy request parameters. 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: ModelCopyAuthorization. The ModelCopyAuthorization is compatible with MutableMapping :rtype: ~azure.ai.documentintelligence.models.ModelCopyAuthorization :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace def authorize_model_copy( self, body: Union[_models.AuthorizeCopyRequest, JSON, IO[bytes]], **kwargs: Any ) -> _models.ModelCopyAuthorization: """Generates authorization to copy a document model to this location with specified modelId and optional description. :param body: Authorize copy request parameters. Is one of the following types: AuthorizeCopyRequest, JSON, IO[bytes] Required. :type body: ~azure.ai.documentintelligence.models.AuthorizeCopyRequest or JSON or IO[bytes] :return: ModelCopyAuthorization. The ModelCopyAuthorization is compatible with MutableMapping :rtype: ~azure.ai.documentintelligence.models.ModelCopyAuthorization :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.ModelCopyAuthorization] = kwargs.pop("cls", 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_document_intelligence_administration_authorize_model_copy_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) _stream = kwargs.pop("stream", False) pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.DocumentIntelligenceErrorResponse, response.json()) raise HttpResponseError(response=response, model=error) if _stream: deserialized = response.iter_bytes() else: deserialized = _deserialize(_models.ModelCopyAuthorization, response.json()) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore def _copy_model_to_initial( self, model_id: str, body: Union[_models.ModelCopyAuthorization, JSON, IO[bytes]], **kwargs: Any ) -> Iterator[bytes]: error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } error_map.update(kwargs.pop("error_map", {}) or {}) _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = kwargs.pop("params", {}) or {} content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) content_type = content_type or "application/json" _content = None if isinstance(body, (IOBase, bytes)): _content = body else: _content = json.dumps(body, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore _request = build_document_intelligence_administration_copy_model_to_request( model_id=model_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) _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) error = _failsafe_deserialize(_models.DocumentIntelligenceErrorResponse, response.json()) raise HttpResponseError(response=response, model=error) response_headers = {} response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) response_headers["Operation-Location"] = self._deserialize("str", response.headers.get("Operation-Location")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @overload def begin_copy_model_to( self, model_id: str, body: _models.ModelCopyAuthorization, *, content_type: str = "application/json", **kwargs: Any, ) -> LROPoller[_models.DocumentModelDetails]: """Copies document model to the target resource, region, and modelId. :param model_id: Unique document model name. Required. :type model_id: str :param body: Copy to request parameters. Required. :type body: ~azure.ai.documentintelligence.models.ModelCopyAuthorization :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 DocumentModelDetails. The DocumentModelDetails is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.DocumentModelDetails] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_copy_model_to( self, model_id: str, body: JSON, *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.DocumentModelDetails]: """Copies document model to the target resource, region, and modelId. :param model_id: Unique document model name. Required. :type model_id: str :param body: Copy to request parameters. 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: An instance of LROPoller that returns DocumentModelDetails. The DocumentModelDetails is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.DocumentModelDetails] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_copy_model_to( self, model_id: str, body: IO[bytes], *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.DocumentModelDetails]: """Copies document model to the target resource, region, and modelId. :param model_id: Unique document model name. Required. :type model_id: str :param body: Copy to request parameters. 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: An instance of LROPoller that returns DocumentModelDetails. The DocumentModelDetails is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.DocumentModelDetails] :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace def begin_copy_model_to( self, model_id: str, body: Union[_models.ModelCopyAuthorization, JSON, IO[bytes]], **kwargs: Any ) -> LROPoller[_models.DocumentModelDetails]: """Copies document model to the target resource, region, and modelId. :param model_id: Unique document model name. Required. :type model_id: str :param body: Copy to request parameters. Is one of the following types: ModelCopyAuthorization, JSON, IO[bytes] Required. :type body: ~azure.ai.documentintelligence.models.ModelCopyAuthorization or JSON or IO[bytes] :return: An instance of LROPoller that returns DocumentModelDetails. The DocumentModelDetails is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.DocumentModelDetails] :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.DocumentModelDetails] = 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_model_to_initial( model_id=model_id, body=body, 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["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) response_headers["Operation-Location"] = self._deserialize( "str", response.headers.get("Operation-Location") ) deserialized = _deserialize(_models.DocumentModelDetails, 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.DocumentModelDetails].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.DocumentModelDetails]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) @distributed_trace def get_model(self, model_id: str, **kwargs: Any) -> _models.DocumentModelDetails: """Gets detailed document model information. :param model_id: Unique document model name. Required. :type model_id: str :return: DocumentModelDetails. The DocumentModelDetails is compatible with MutableMapping :rtype: ~azure.ai.documentintelligence.models.DocumentModelDetails :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.DocumentModelDetails] = kwargs.pop("cls", None) _request = build_document_intelligence_administration_get_model_request( model_id=model_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 = kwargs.pop("stream", False) pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.DocumentIntelligenceErrorResponse, response.json()) raise HttpResponseError(response=response, model=error) 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() else: deserialized = _deserialize(_models.DocumentModelDetails, response.json()) if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @distributed_trace def list_models(self, **kwargs: Any) -> Iterable["_models.DocumentModelDetails"]: """List all document models. :return: An iterator like instance of DocumentModelDetails :rtype: ~azure.core.paging.ItemPaged[~azure.ai.documentintelligence.models.DocumentModelDetails] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[List[_models.DocumentModelDetails]] = 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_document_intelligence_administration_list_models_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.DocumentModelDetails], deserialized["value"]) if cls: list_of_elem = cls(list_of_elem) # type: ignore return deserialized.get("nextLink") or None, iter(list_of_elem) def get_next(next_link=None): _request = prepare_request(next_link) _stream = False pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.DocumentIntelligenceErrorResponse, response.json()) raise HttpResponseError(response=response, model=error) return pipeline_response return ItemPaged(get_next, extract_data) @distributed_trace def delete_model(self, model_id: str, **kwargs: Any) -> None: # pylint: disable=inconsistent-return-statements """Deletes document model. :param model_id: Unique document model name. Required. :type model_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_document_intelligence_administration_delete_model_request( model_id=model_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) error = _failsafe_deserialize(_models.DocumentIntelligenceErrorResponse, response.json()) raise HttpResponseError(response=response, model=error) 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 get_resource_details(self, **kwargs: Any) -> _models.DocumentIntelligenceResourceDetails: """Return information about the current resource. :return: DocumentIntelligenceResourceDetails. The DocumentIntelligenceResourceDetails is compatible with MutableMapping :rtype: ~azure.ai.documentintelligence.models.DocumentIntelligenceResourceDetails :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.DocumentIntelligenceResourceDetails] = kwargs.pop("cls", None) _request = build_document_intelligence_administration_get_resource_details_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) _stream = kwargs.pop("stream", False) pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.DocumentIntelligenceErrorResponse, response.json()) raise HttpResponseError(response=response, model=error) if _stream: deserialized = response.iter_bytes() else: deserialized = _deserialize(_models.DocumentIntelligenceResourceDetails, response.json()) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore @distributed_trace def get_operation(self, operation_id: str, **kwargs: Any) -> _models.DocumentIntelligenceOperationDetails: """Gets operation info. :param operation_id: Operation ID. Required. :type operation_id: str :return: DocumentIntelligenceOperationDetails. The DocumentIntelligenceOperationDetails is compatible with MutableMapping :rtype: ~azure.ai.documentintelligence.models.DocumentIntelligenceOperationDetails :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.DocumentIntelligenceOperationDetails] = kwargs.pop("cls", None) _request = build_document_intelligence_administration_get_operation_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 = kwargs.pop("stream", False) pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.DocumentIntelligenceErrorResponse, response.json()) raise HttpResponseError(response=response, model=error) 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() else: deserialized = _deserialize(_models.DocumentIntelligenceOperationDetails, response.json()) if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @distributed_trace def list_operations(self, **kwargs: Any) -> Iterable["_models.DocumentIntelligenceOperationDetails"]: """Lists all operations. :return: An iterator like instance of DocumentIntelligenceOperationDetails :rtype: ~azure.core.paging.ItemPaged[~azure.ai.documentintelligence.models.DocumentIntelligenceOperationDetails] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[List[_models.DocumentIntelligenceOperationDetails]] = 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_document_intelligence_administration_list_operations_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.DocumentIntelligenceOperationDetails], deserialized["value"]) if cls: list_of_elem = cls(list_of_elem) # type: ignore return deserialized.get("nextLink") or None, iter(list_of_elem) def get_next(next_link=None): _request = prepare_request(next_link) _stream = False pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.DocumentIntelligenceErrorResponse, response.json()) raise HttpResponseError(response=response, model=error) return pipeline_response return ItemPaged(get_next, extract_data) def _build_classifier_initial( self, body: Union[_models.BuildDocumentClassifierRequest, JSON, IO[bytes]], **kwargs: Any ) -> Iterator[bytes]: error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } error_map.update(kwargs.pop("error_map", {}) or {}) _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = kwargs.pop("params", {}) or {} content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) content_type = content_type or "application/json" _content = None if isinstance(body, (IOBase, bytes)): _content = body else: _content = json.dumps(body, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore _request = build_document_intelligence_administration_build_classifier_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) _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) error = _failsafe_deserialize(_models.DocumentIntelligenceErrorResponse, response.json()) raise HttpResponseError(response=response, model=error) response_headers = {} response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) response_headers["Operation-Location"] = self._deserialize("str", response.headers.get("Operation-Location")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @overload def begin_build_classifier( self, body: _models.BuildDocumentClassifierRequest, *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.DocumentClassifierDetails]: """Builds a custom document classifier. :param body: Build request parameters. Required. :type body: ~azure.ai.documentintelligence.models.BuildDocumentClassifierRequest :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 DocumentClassifierDetails. The DocumentClassifierDetails is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.DocumentClassifierDetails] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_build_classifier( self, body: JSON, *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.DocumentClassifierDetails]: """Builds a custom document classifier. :param body: Build request parameters. 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: An instance of LROPoller that returns DocumentClassifierDetails. The DocumentClassifierDetails is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.DocumentClassifierDetails] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_build_classifier( self, body: IO[bytes], *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.DocumentClassifierDetails]: """Builds a custom document classifier. :param body: Build request parameters. 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: An instance of LROPoller that returns DocumentClassifierDetails. The DocumentClassifierDetails is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.DocumentClassifierDetails] :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace def begin_build_classifier( self, body: Union[_models.BuildDocumentClassifierRequest, JSON, IO[bytes]], **kwargs: Any ) -> LROPoller[_models.DocumentClassifierDetails]: """Builds a custom document classifier. :param body: Build request parameters. Is one of the following types: BuildDocumentClassifierRequest, JSON, IO[bytes] Required. :type body: ~azure.ai.documentintelligence.models.BuildDocumentClassifierRequest or JSON or IO[bytes] :return: An instance of LROPoller that returns DocumentClassifierDetails. The DocumentClassifierDetails is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.DocumentClassifierDetails] :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.DocumentClassifierDetails] = 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._build_classifier_initial( body=body, 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["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) response_headers["Operation-Location"] = self._deserialize( "str", response.headers.get("Operation-Location") ) deserialized = _deserialize(_models.DocumentClassifierDetails, 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.DocumentClassifierDetails].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.DocumentClassifierDetails]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) @overload def authorize_classifier_copy( self, body: _models.AuthorizeClassifierCopyRequest, *, content_type: str = "application/json", **kwargs: Any ) -> _models.ClassifierCopyAuthorization: """Generates authorization to copy a document classifier to this location with specified classifierId and optional description. :param body: Authorize copy request parameters. Required. :type body: ~azure.ai.documentintelligence.models.AuthorizeClassifierCopyRequest :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :return: ClassifierCopyAuthorization. The ClassifierCopyAuthorization is compatible with MutableMapping :rtype: ~azure.ai.documentintelligence.models.ClassifierCopyAuthorization :raises ~azure.core.exceptions.HttpResponseError: """ @overload def authorize_classifier_copy( self, body: JSON, *, content_type: str = "application/json", **kwargs: Any ) -> _models.ClassifierCopyAuthorization: """Generates authorization to copy a document classifier to this location with specified classifierId and optional description. :param body: Authorize copy request parameters. 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: ClassifierCopyAuthorization. The ClassifierCopyAuthorization is compatible with MutableMapping :rtype: ~azure.ai.documentintelligence.models.ClassifierCopyAuthorization :raises ~azure.core.exceptions.HttpResponseError: """ @overload def authorize_classifier_copy( self, body: IO[bytes], *, content_type: str = "application/json", **kwargs: Any ) -> _models.ClassifierCopyAuthorization: """Generates authorization to copy a document classifier to this location with specified classifierId and optional description. :param body: Authorize copy request parameters. 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: ClassifierCopyAuthorization. The ClassifierCopyAuthorization is compatible with MutableMapping :rtype: ~azure.ai.documentintelligence.models.ClassifierCopyAuthorization :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace def authorize_classifier_copy( self, body: Union[_models.AuthorizeClassifierCopyRequest, JSON, IO[bytes]], **kwargs: Any ) -> _models.ClassifierCopyAuthorization: """Generates authorization to copy a document classifier to this location with specified classifierId and optional description. :param body: Authorize copy request parameters. Is one of the following types: AuthorizeClassifierCopyRequest, JSON, IO[bytes] Required. :type body: ~azure.ai.documentintelligence.models.AuthorizeClassifierCopyRequest or JSON or IO[bytes] :return: ClassifierCopyAuthorization. The ClassifierCopyAuthorization is compatible with MutableMapping :rtype: ~azure.ai.documentintelligence.models.ClassifierCopyAuthorization :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.ClassifierCopyAuthorization] = kwargs.pop("cls", 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_document_intelligence_administration_authorize_classifier_copy_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) _stream = kwargs.pop("stream", False) pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.DocumentIntelligenceErrorResponse, response.json()) raise HttpResponseError(response=response, model=error) if _stream: deserialized = response.iter_bytes() else: deserialized = _deserialize(_models.ClassifierCopyAuthorization, response.json()) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore def _copy_classifier_to_initial( self, classifier_id: str, body: Union[_models.ClassifierCopyAuthorization, JSON, IO[bytes]], **kwargs: Any ) -> Iterator[bytes]: error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } error_map.update(kwargs.pop("error_map", {}) or {}) _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = kwargs.pop("params", {}) or {} content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) content_type = content_type or "application/json" _content = None if isinstance(body, (IOBase, bytes)): _content = body else: _content = json.dumps(body, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore _request = build_document_intelligence_administration_copy_classifier_to_request( classifier_id=classifier_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) _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) error = _failsafe_deserialize(_models.DocumentIntelligenceErrorResponse, response.json()) raise HttpResponseError(response=response, model=error) response_headers = {} response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) response_headers["Operation-Location"] = self._deserialize("str", response.headers.get("Operation-Location")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @overload def begin_copy_classifier_to( self, classifier_id: str, body: _models.ClassifierCopyAuthorization, *, content_type: str = "application/json", **kwargs: Any, ) -> LROPoller[_models.DocumentClassifierDetails]: """Copies document classifier to the target resource, region, and classifierId. :param classifier_id: Unique document classifier name. Required. :type classifier_id: str :param body: Copy to request parameters. Required. :type body: ~azure.ai.documentintelligence.models.ClassifierCopyAuthorization :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 DocumentClassifierDetails. The DocumentClassifierDetails is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.DocumentClassifierDetails] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_copy_classifier_to( self, classifier_id: str, body: JSON, *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.DocumentClassifierDetails]: """Copies document classifier to the target resource, region, and classifierId. :param classifier_id: Unique document classifier name. Required. :type classifier_id: str :param body: Copy to request parameters. 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: An instance of LROPoller that returns DocumentClassifierDetails. The DocumentClassifierDetails is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.DocumentClassifierDetails] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_copy_classifier_to( self, classifier_id: str, body: IO[bytes], *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.DocumentClassifierDetails]: """Copies document classifier to the target resource, region, and classifierId. :param classifier_id: Unique document classifier name. Required. :type classifier_id: str :param body: Copy to request parameters. 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: An instance of LROPoller that returns DocumentClassifierDetails. The DocumentClassifierDetails is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.DocumentClassifierDetails] :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace def begin_copy_classifier_to( self, classifier_id: str, body: Union[_models.ClassifierCopyAuthorization, JSON, IO[bytes]], **kwargs: Any ) -> LROPoller[_models.DocumentClassifierDetails]: """Copies document classifier to the target resource, region, and classifierId. :param classifier_id: Unique document classifier name. Required. :type classifier_id: str :param body: Copy to request parameters. Is one of the following types: ClassifierCopyAuthorization, JSON, IO[bytes] Required. :type body: ~azure.ai.documentintelligence.models.ClassifierCopyAuthorization or JSON or IO[bytes] :return: An instance of LROPoller that returns DocumentClassifierDetails. The DocumentClassifierDetails is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.ai.documentintelligence.models.DocumentClassifierDetails] :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.DocumentClassifierDetails] = 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_classifier_to_initial( classifier_id=classifier_id, body=body, 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["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) response_headers["Operation-Location"] = self._deserialize( "str", response.headers.get("Operation-Location") ) deserialized = _deserialize(_models.DocumentClassifierDetails, 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.DocumentClassifierDetails].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.DocumentClassifierDetails]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) @distributed_trace def get_classifier(self, classifier_id: str, **kwargs: Any) -> _models.DocumentClassifierDetails: """Gets detailed document classifier information. :param classifier_id: Unique document classifier name. Required. :type classifier_id: str :return: DocumentClassifierDetails. The DocumentClassifierDetails is compatible with MutableMapping :rtype: ~azure.ai.documentintelligence.models.DocumentClassifierDetails :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.DocumentClassifierDetails] = kwargs.pop("cls", None) _request = build_document_intelligence_administration_get_classifier_request( classifier_id=classifier_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 = kwargs.pop("stream", False) pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.DocumentIntelligenceErrorResponse, response.json()) raise HttpResponseError(response=response, model=error) 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() else: deserialized = _deserialize(_models.DocumentClassifierDetails, response.json()) if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @distributed_trace def list_classifiers(self, **kwargs: Any) -> Iterable["_models.DocumentClassifierDetails"]: """List all document classifiers. :return: An iterator like instance of DocumentClassifierDetails :rtype: ~azure.core.paging.ItemPaged[~azure.ai.documentintelligence.models.DocumentClassifierDetails] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[List[_models.DocumentClassifierDetails]] = 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_document_intelligence_administration_list_classifiers_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.DocumentClassifierDetails], deserialized["value"]) if cls: list_of_elem = cls(list_of_elem) # type: ignore return deserialized.get("nextLink") or None, iter(list_of_elem) def get_next(next_link=None): _request = prepare_request(next_link) _stream = False pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.DocumentIntelligenceErrorResponse, response.json()) raise HttpResponseError(response=response, model=error) return pipeline_response return ItemPaged(get_next, extract_data) @distributed_trace def delete_classifier( # pylint: disable=inconsistent-return-statements self, classifier_id: str, **kwargs: Any ) -> None: """Deletes document classifier. :param classifier_id: Unique document classifier name. Required. :type classifier_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_document_intelligence_administration_delete_classifier_request( classifier_id=classifier_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) error = _failsafe_deserialize(_models.DocumentIntelligenceErrorResponse, response.json()) raise HttpResponseError(response=response, model=error) 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