File: _patch.py

package info (click to toggle)
python-azure 20230112%2Bgit-1
  • links: PTS, VCS
  • area: main
  • in suites: bookworm
  • size: 749,544 kB
  • sloc: python: 6,815,827; javascript: 287; makefile: 195; xml: 109; sh: 105
file content (274 lines) | stat: -rw-r--r-- 12,245 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
# ------------------------------------
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
# ------------------------------------
"""Customize generated code here.

Follow our quickstart for examples: https://aka.ms/azsdk/python/dpcodegen/python/customize
"""
from typing import Any, List, overload, Optional, Union, Tuple, cast, MutableMapping
import copy
from azure.core.tracing.decorator import distributed_trace

from ._operations import QuestionAnsweringClientOperationsMixin as QuestionAnsweringClientOperationsMixinGenerated
from ..models import (
    AnswersOptions,
    AnswersFromTextOptions,
    AnswersResult,
    AnswersFromTextResult,
    KnowledgeBaseAnswerContext,
    QueryFilters,
    ShortAnswerOptions,
    TextDocument,
)

JSON = MutableMapping[str, Any]


def _validate_text_records(records):
    if not records:
        raise ValueError("Input documents can not be empty or None")
    if isinstance(records, str):
        raise TypeError("Input documents cannot be a string.")
    if isinstance(records, dict):
        raise TypeError("Input documents cannot be a dict")
    if not all(isinstance(x, str) for x in records):
        if not all(isinstance(x, (dict, TextDocument)) for x in records):
            raise TypeError("Mixing string and dictionary/object document input unsupported.")
    request_batch = []
    for idx, doc in enumerate(records):
        if isinstance(doc, str):
            record = {"id": str(idx), "text": doc}
            request_batch.append(record)
        else:
            request_batch.append(doc)
    return request_batch


def _get_positional_body(*args, **kwargs):
    """Verify args and kwargs are valid, and then return the positional body, if users passed it in."""
    if len(args) > 1:
        raise TypeError("There can only be one positional argument, which is the POST body of this request.")
    if "options" in kwargs:
        raise TypeError("The 'options' parameter is positional only.")
    return args[0] if args else None


def _verify_qna_id_and_question(query_knowledgebase_options):
    """For query_knowledge_base we require either `question` or `qna_id`."""
    try:
        qna_id = query_knowledgebase_options.qna_id
        question = query_knowledgebase_options.question
    except AttributeError:
        qna_id = query_knowledgebase_options.get("qna_id") or query_knowledgebase_options.get("qnaId")
        question = query_knowledgebase_options.get("question")
    if not (qna_id or question):
        raise TypeError("You need to pass in either `qna_id` or `question`.")
    if qna_id and question:
        raise TypeError("You can not specify both `qna_id` and `question`.")


def _handle_metadata_filter_conversion(options_input):
    options = copy.deepcopy(options_input)
    filters = options.filters if hasattr(options, "filters") else options.get("filters", {})
    try:
        if filters and filters.metadata_filter and filters.metadata_filter.metadata:
            metadata_input = filters.metadata_filter.metadata
        else:
            metadata_input = None
        in_class = True
    except AttributeError:
        metadata_input = filters.get("metadataFilter", {}).get("metadata")
        in_class = False
    if not metadata_input:
        return options
    try:
        if any(t for t in metadata_input if len(t) != 2):
            raise ValueError("'metadata' must be a sequence of key-value tuples.")
    except TypeError:
        raise ValueError("'metadata' must be a sequence of key-value tuples.")
    metadata_modified = [{"key": m[0], "value": m[1]} for m in metadata_input]
    if in_class:
        filters.metadata_filter.metadata = metadata_modified
    else:
        filters["metadataFilter"]["metadata"] = metadata_modified
    return options


def _get_answers_prepare_options(*args: AnswersOptions, **kwargs: Any) -> Tuple[AnswersOptions, Any]:
    options = _get_positional_body(*args, **kwargs) or AnswersOptions(
        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_threshold=kwargs.pop("confidence_threshold", None),
        answer_context=kwargs.pop("answer_context", None),
        ranker_kind=kwargs.pop("ranker_kind", None),
        filters=kwargs.pop("filters", None),
        short_answer_options=kwargs.pop("short_answer_options", None),
        include_unstructured_sources=kwargs.pop("include_unstructured_sources", None),
    )
    _verify_qna_id_and_question(options)
    return _handle_metadata_filter_conversion(options), kwargs


def _get_answers_from_text_prepare_options(
    *args: AnswersFromTextOptions, **kwargs: Any
) -> Tuple[Union[JSON, AnswersFromTextOptions], Any]:
    default_language = kwargs.pop("language", None)
    options = _get_positional_body(*args, **kwargs) or AnswersFromTextOptions(
        question=kwargs.pop("question"),
        text_documents=kwargs.pop("text_documents"),
        language=default_language,
    )
    try:
        options = cast(JSON, options)
        # pylint: disable=unsubscriptable-object,unsupported-assignment-operation
        options["records"] = _validate_text_records(options["records"])
        # pylint: disable=no-member,unsupported-assignment-operation
        options["language"] = options.get("language", None) or default_language
    except TypeError:
        options = cast(AnswersFromTextOptions, options)
        options.text_documents = _validate_text_records(options.text_documents)
        options.language = options.language or default_language
    return options, kwargs


class QuestionAnsweringClientOperationsMixin(QuestionAnsweringClientOperationsMixinGenerated):
    @overload  # type: ignore # https://github.com/Azure/azure-sdk-for-python/issues/26621
    def get_answers(
        self, options: AnswersOptions, *, project_name: str, deployment_name: str, **kwargs: Any
    ) -> AnswersResult:
        ...

    @overload
    def get_answers(  # pylint: disable=arguments-differ
        self,
        *,
        project_name: str,
        deployment_name: str,
        qna_id: Optional[int] = None,
        question: Optional[str] = None,
        top: Optional[int] = None,
        user_id: Optional[str] = None,
        confidence_threshold: Optional[float] = None,
        answer_context: Optional[KnowledgeBaseAnswerContext] = None,
        ranker_kind: Optional[str] = None,
        filters: Optional[QueryFilters] = None,
        short_answer_options: Optional[ShortAnswerOptions] = None,
        include_unstructured_sources: Optional[bool] = None,
        **kwargs: Any
    ) -> AnswersResult:
        ...

    @distributed_trace
    def get_answers(self, *args: AnswersOptions, **kwargs: Any) -> AnswersResult:
        """Answers the specified question using your knowledge base.

        :param options: Positional only. 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.AnswersOptions
        :keyword project_name: The name of the knowledge base 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 qna_id: Exact QnA ID to fetch from the knowledge base, this field takes priority over
         question.
        :paramtype qna_id: int
        :keyword question: User question to query against the knowledge base.
        :paramtype question: str
        :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_threshold: Minimum threshold score for answers, value ranges from 0 to 1.
        :paramtype confidence_threshold: float
        :keyword answer_context: Context object with previous QnA's information.
        :paramtype answer_context: ~azure.ai.language.questionanswering.models.KnowledgeBaseAnswerContext
        :keyword ranker_kind: Type of ranker to be used. Possible
         values include: "Default", "QuestionOnly".
        :paramtype ranker_kind: str
        :keyword filters: Filter QnAs based on given metadata list and knowledge base sources.
        :paramtype filters: ~azure.ai.language.questionanswering.models.QueryFilters
        :keyword short_answer_options: To configure Answer span prediction feature.
        :paramtype short_answer_options: ~azure.ai.language.questionanswering.models.ShortAnswerOptions
        :keyword include_unstructured_sources: (Optional) Flag to enable Query over Unstructured
         Sources.
        :paramtype include_unstructured_sources: bool
        :return: AnswersResult
        :rtype: ~azure.ai.language.questionanswering.models.AnswersResult
        :raises ~azure.core.exceptions.HttpResponseError:

        .. admonition:: Example:

            .. literalinclude:: ../samples/sample_query_knowledgebase.py
                :start-after: [START query_knowledgebase]
                :end-before: [END query_knowledgebase]
                :language: python
                :dedent: 4
                :caption: Answer the specified question using your knowledge base.
        """
        options, kwargs = _get_answers_prepare_options(*args, **kwargs)
        return super().get_answers(options, **kwargs)

    @overload  # type: ignore
    def get_answers_from_text(self, options: AnswersFromTextOptions, **kwargs: Any) -> AnswersFromTextResult:
        pass

    @overload
    def get_answers_from_text(  # pylint: disable=arguments-differ
        self,
        *,
        question: str,
        text_documents: List[Union[str, TextDocument]],
        language: Optional[str] = None,
        **kwargs: Any
    ) -> AnswersFromTextResult:
        ...

    @distributed_trace
    def get_answers_from_text(self, *args: AnswersFromTextOptions, **kwargs: Any) -> AnswersFromTextResult:
        """Answers the specified question using the provided text in the body.

        :param options: Positional only. 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.AnswersFromTextOptions
        :keyword question: User question to query against the given text records.
        :paramtype question: str
        :keyword text_documents: Text records to be searched for given question.
        :paramtype text_documents: list[str or ~azure.ai.language.questionanswering.models.TextDocument]
        :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: AnswersFromTextResult
        :rtype: ~azure.ai.language.questionanswering.models.AnswersFromTextResult
        :raises ~azure.core.exceptions.HttpResponseError:

        .. admonition:: Example:

            .. literalinclude:: ../samples/sample_query_text.py
                :start-after: [START query_text]
                :end-before: [END query_text]
                :language: python
                :dedent: 4
                :caption: Answers the specified question using the provided text.
        """
        options, kwargs = _get_answers_from_text_prepare_options(
            *args, language=kwargs.pop("language", self._default_language), **kwargs  # type: ignore
        )
        return super().get_answers_from_text(options, **kwargs)  # type: ignore


__all__: List[str] = [
    "QuestionAnsweringClientOperationsMixin"
]  # Add all objects you want publicly available to users at this package level


def patch_sdk():
    """Do not remove from this file.

    `patch_sdk` is a last resort escape hatch that allows you to do customizations
    you can't accomplish using the techniques described in
    https://aka.ms/azsdk/python/dpcodegen/python/customize
    """