Source code for azure.ai.language.questionanswering.operations._operations

# coding=utf-8
# --------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is regenerated.
# --------------------------------------------------------------------------
import functools
from typing import TYPE_CHECKING, overload
import warnings

from azure.core.exceptions import (
    ClientAuthenticationError,
    HttpResponseError,
    ResourceExistsError,
    ResourceNotFoundError,
    map_error,
)
from azure.core.pipeline import PipelineResponse
from azure.core.pipeline.transport import HttpResponse
from azure.core.rest import HttpRequest
from msrest import Serializer

from .. import models as _models
from .._patch import _validate_text_records, _get_positional_body, _verify_qna_id_and_question

if TYPE_CHECKING:
    # pylint: disable=unused-import,ungrouped-imports
    from typing import Any, Callable, Dict, Generic, Optional, TypeVar

    T = TypeVar("T")
    ClsType = Optional[Callable[[PipelineResponse[HttpRequest, HttpResponse], T, Dict[str, Any]], Any]]

_SERIALIZER = Serializer()
# fmt: off

def build_query_knowledge_base_request(
    **kwargs  # type: Any
):
    # type: (...) -> HttpRequest
    content_type = kwargs.pop('content_type', None)  # type: Optional[str]
    project_name = kwargs.pop('project_name')  # type: str
    deployment_name = kwargs.pop('deployment_name')  # type: str

    api_version = "2021-07-15-preview"
    accept = "application/json"
    # Construct URL
    url = kwargs.pop("template_url", '/:query-knowledgebases')

    # Construct parameters
    query_parameters = kwargs.pop("params", {})  # type: Dict[str, Any]
    query_parameters['projectName'] = _SERIALIZER.query("project_name", project_name, 'str')
    query_parameters['deploymentName'] = _SERIALIZER.query("deployment_name", deployment_name, 'str')
    query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')

    # Construct headers
    header_parameters = kwargs.pop("headers", {})  # type: Dict[str, Any]
    if content_type is not None:
        header_parameters['Content-Type'] = _SERIALIZER.header("content_type", content_type, 'str')
    header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')

    return HttpRequest(
        method="POST",
        url=url,
        params=query_parameters,
        headers=header_parameters,
        **kwargs
    )


def build_query_text_request(
    **kwargs  # type: Any
):
    # type: (...) -> HttpRequest
    content_type = kwargs.pop('content_type', None)  # type: Optional[str]

    api_version = "2021-07-15-preview"
    accept = "application/json"
    # Construct URL
    url = kwargs.pop("template_url", '/:query-text')

    # Construct parameters
    query_parameters = kwargs.pop("params", {})  # type: Dict[str, Any]
    query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')

    # Construct headers
    header_parameters = kwargs.pop("headers", {})  # type: Dict[str, Any]
    if content_type is not None:
        header_parameters['Content-Type'] = _SERIALIZER.header("content_type", content_type, 'str')
    header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')

    return HttpRequest(
        method="POST",
        url=url,
        params=query_parameters,
        headers=header_parameters,
        **kwargs
    )

# fmt: on
[docs]class QuestionAnsweringClientOperationsMixin(object): @overload def query_knowledge_base( self, options, # type: "_models.QueryKnowledgeBaseOptions" **kwargs # type: Any ): # type: (...) -> "_models.KnowledgeBaseAnswers" """Answers the specified question using your knowledge base. :param options: Positional-only POST body of the request. :type options: ~azure.ai.language.questionanswering.models.QueryKnowledgeBaseOptions :keyword project_name: The name of the project to use. :paramtype project_name: str :keyword deployment_name: The name of the specific deployment of the project to use. :paramtype deployment_name: str :return: KnowledgeBaseAnswers :rtype: ~azure.ai.language.questionanswering.models.KnowledgeBaseAnswers :raises: ~azure.core.exceptions.HttpResponseError """ pass @overload def query_knowledge_base( self, **kwargs # type: Any ): # type: (...) -> "_models.KnowledgeBaseAnswers" """Answers the specified question using your knowledge base. :keyword project_name: The name of the project to use. :paramtype project_name: str :keyword deployment_name: The name of the specific deployment of the project to use. :paramtype deployment_name: str :keyword question: User question to query against the knowledge base. :paramtype question: str :keyword qna_id: Exact QnA ID to fetch from the knowledge base, this field takes priority over question. :paramtype qna_id: int :keyword top: Max number of answers to be returned for the question. :paramtype top: int :keyword user_id: Unique identifier for the user. :paramtype user_id: str :keyword confidence_score_threshold: Minimum threshold score for answers, value ranges from 0 to 1. :paramtype confidence_score_threshold: float :keyword context: Context object with previous QnA's information. :paramtype context: ~azure.ai.language.questionanswering.models.KnowledgeBaseAnswerRequestContext :keyword ranker_type: (Optional) Set to 'QuestionOnly' for using a question only Ranker. Possible values include: "Default", "QuestionOnly". :paramtype ranker_type: str or ~azure.ai.language.questionanswering.models.RankerType :keyword filters: Filter QnAs based on give metadata list and knowledge base source names. :paramtype filters: ~azure.ai.language.questionanswering.models.QueryFilters :keyword answer_span_request: To configure Answer span prediction feature. :paramtype answer_span_request: ~azure.ai.language.questionanswering.models.AnswerSpanRequest :keyword include_unstructured_sources: (Optional) Flag to enable Query over Unstructured Sources. :paramtype include_unstructured_sources: bool :return: KnowledgeBaseAnswers :rtype: ~azure.ai.language.questionanswering.models.KnowledgeBaseAnswers :raises: ~azure.core.exceptions.HttpResponseError """ pass
[docs] def query_knowledge_base( self, *args, # type: "_models.QueryKnowledgeBaseOptions" **kwargs # type: Any ): # type: (...) -> "_models.KnowledgeBaseAnswers" """Answers the specified question using your knowledge base. :param options: POST body of the request. Provide either `options`, OR individual keyword arguments. If both are provided, only the options object will be used. :type options: ~azure.ai.language.questionanswering.models.QueryKnowledgeBaseOptions :keyword project_name: The name of the project to use. :paramtype project_name: str :keyword deployment_name: The name of the specific deployment of the project to use. :paramtype deployment_name: str :keyword question: User question to query against the knowledge base. Provide either `options`, OR individual keyword arguments. If both are provided, only the options object will be used. :paramtype question: str :keyword qna_id: Exact QnA ID to fetch from the knowledge base, this field takes priority over question. :paramtype qna_id: int :keyword top: Max number of answers to be returned for the question. :paramtype top: int :keyword user_id: Unique identifier for the user. :paramtype user_id: str :keyword confidence_score_threshold: Minimum threshold score for answers, value ranges from 0 to 1. :paramtype confidence_score_threshold: float :keyword context: Context object with previous QnA's information. :paramtype context: ~azure.ai.language.questionanswering.models.KnowledgeBaseAnswerRequestContext :keyword ranker_type: (Optional) Set to 'QuestionOnly' for using a question only Ranker. Possible values include: "Default", "QuestionOnly". :paramtype ranker_type: str or ~azure.ai.language.questionanswering.models.RankerType :keyword filters: Filter QnAs based on give metadata list and knowledge base source names. :paramtype filters: ~azure.ai.language.questionanswering.models.QueryFilters :keyword answer_span_request: To configure Answer span prediction feature. :paramtype answer_span_request: ~azure.ai.language.questionanswering.models.AnswerSpanRequest :keyword include_unstructured_sources: (Optional) Flag to enable Query over Unstructured Sources. :paramtype include_unstructured_sources: bool :return: KnowledgeBaseAnswers :rtype: ~azure.ai.language.questionanswering.models.KnowledgeBaseAnswers :raises: ~azure.core.exceptions.HttpResponseError """ options = _get_positional_body(*args, **kwargs) or _models.QueryKnowledgeBaseOptions( qna_id=kwargs.pop("qna_id", None), question=kwargs.pop("question", None), top=kwargs.pop("top", None), user_id=kwargs.pop("user_id", None), confidence_score_threshold=kwargs.pop("confidence_score_threshold", None), context=kwargs.pop("context", None), ranker_type=kwargs.pop("ranker_type", None), filters=kwargs.pop("filters", None), answer_span_request=kwargs.pop("answer_span_request", None), include_unstructured_sources=kwargs.pop("include_unstructured_sources", None) ) _verify_qna_id_and_question(options) cls = kwargs.pop("cls", None) # type: ClsType["_models.KnowledgeBaseAnswers"] error_map = {401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError} error_map.update(kwargs.pop("error_map", {})) content_type = kwargs.pop("content_type", "application/json") # type: Optional[str] project_name = kwargs.pop("project_name") # type: str deployment_name = kwargs.pop("deployment_name") # type: str json = self._serialize.body(options, "QueryKnowledgeBaseOptions") request = build_query_knowledge_base_request( content_type=content_type, project_name=project_name, deployment_name=deployment_name, json=json, template_url=self.query_knowledge_base.metadata["url"], ) path_format_arguments = { "Endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, "str", skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize("KnowledgeBaseAnswers", pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized
query_knowledge_base.metadata = {"url": "/:query-knowledgebases"} # type: ignore @overload def query_text( self, options, # type: "_models.QueryTextOptions" **kwargs # type: Any ): # type: (...) -> "_models.TextAnswers" """Answers the specified question using the provided text in the body. :param options: Positional-only POST body of the request. :type options: ~azure.ai.language.questionanswering.models.QueryTextOptions :return: TextAnswers :rtype: ~azure.ai.language.questionanswering.models.TextAnswers :raises: ~azure.core.exceptions.HttpResponseError """ pass @overload def query_text( self, **kwargs # type: Any ): # type: (...) -> "_models.TextAnswers" """Answers the specified question using the provided text in the body. :keyword question: Required. User question to query against the given text records. :paramtype question: str :keyword records: Required. Text records to be searched for given question. :paramtype records: list[str or ~azure.ai.language.questionanswering.models.TextRecord] :keyword language: Language of the text records. This is BCP-47 representation of a language. For example, use "en" for English; "es" for Spanish etc. If not set, use "en" for English as default. :paramtype language: str :return: TextAnswers :rtype: ~azure.ai.language.questionanswering.models.TextAnswers :raises: ~azure.core.exceptions.HttpResponseError """ pass
[docs] def query_text( self, *args, # type: "_models.QueryTextOptions" **kwargs # type: Any ): # type: (...) -> "_models.TextAnswers" """Answers the specified question using the provided text in the body. :param options: POST body of the request. Provide either `options`, OR individual keyword arguments. If both are provided, only the options object will be used. :type options: ~azure.ai.language.questionanswering.models.QueryTextOptions :keyword question: User question to query against the given text records. Provide either `options`, OR individual keyword arguments. If both are provided, only the options object will be used. :paramtype question: str :keyword records: Text records to be searched for given question. Provide either `options`, OR individual keyword arguments. If both are provided, only the options object will be used. :paramtype records: list[str or ~azure.ai.language.questionanswering.models.TextRecord] :keyword language: Language of the text records. This is BCP-47 representation of a language. For example, use "en" for English; "es" for Spanish etc. If not set, use "en" for English as default. :paramtype language: str :return: TextAnswers :rtype: ~azure.ai.language.questionanswering.models.TextAnswers :raises: ~azure.core.exceptions.HttpResponseError """ options = _get_positional_body(*args, **kwargs) or _models.QueryTextOptions( question=kwargs.pop("question"), records=kwargs.pop("records"), language=kwargs.pop("language", self._default_language), ) try: options['records'] = _validate_text_records(options['records']) except TypeError: options.records = _validate_text_records(options.records) cls = kwargs.pop("cls", None) # type: ClsType["_models.TextAnswers"] error_map = {401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError} error_map.update(kwargs.pop("error_map", {})) content_type = kwargs.pop("content_type", "application/json") # type: Optional[str] json = self._serialize.body(options, "QueryTextOptions") request = build_query_text_request( content_type=content_type, json=json, template_url=self.query_text.metadata["url"], ) path_format_arguments = { "Endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, "str", skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize("TextAnswers", pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized
query_text.metadata = {"url": "/:query-text"} # type: ignore