如何在Python API for Azure DevOps 7.0+版本中检索延续令牌?

hs1rzwqc  于 2023-05-21  发布在  Python
关注(0)|答案(1)|浏览(159)

bounty已结束。回答此问题可获得+100声望奖励。赏金宽限期17小时后结束。Jean-Francois T.正在寻找规范答案:请为这个问题提供一个官方的答案,而不仅仅是一个(脏)补丁。

在早期版本的Azure DevOps Python API(6.0.0b4之前)中,当对某些项目(例如WorkItems,Test Suites,...),您有一个带有valuecontinuation_token的响应对象,您可以使用它来发出新的请求并继续解析。
例如,下面是这样的函数的原型:

def get_test_suites_for_plan(self, project, plan_id, expand=None, continuation_token=None, as_tree_view=None):
        """GetTestSuitesForPlan.
        [Preview API] Get test suites for plan.
        :param str project: Project ID or project name
        :param int plan_id: ID of the test plan for which suites are requested.
        :param str expand: Include the children suites and testers details.
        :param str continuation_token: If the list of suites returned is not complete, a continuation token to query next batch of suites is included in the response header as "x-ms-continuationtoken". Omit this parameter to get the first batch of test suites.
        :param bool as_tree_view: If the suites returned should be in a tree structure.
        :rtype: :class:`<GetTestSuitesForPlanResponseValue>`

所以你可以这样做:

resp = client.get_test_suites_for_plan(project, my_plan_id)
suites = resp.value
while resp.continuation_token:
    resp = client.get_test_suites_for_plan(project, my_plan_id)
    suites += resp.value

在最近的版本(特别是7.0)中,现在返回的是一个列表(但受API的大小限制)。
例如,类似函数的版本将是:

def get_test_suites_for_plan(self, project, plan_id, expand=None, continuation_token=None, as_tree_view=None):
        """GetTestSuitesForPlan.
        [Preview API] Get test suites for plan.
        :param str project: Project ID or project name
        :param int plan_id: ID of the test plan for which suites are requested.
        :param str expand: Include the children suites and testers details.
        :param str continuation_token: If the list of suites returned is not complete, a continuation token to query next batch of suites is included in the response header as "x-ms-continuationtoken". Omit this parameter to get the first batch of test suites.
        :param bool as_tree_view: If the suites returned should be in a tree structure.
        :rtype: :class:`<[TestSuite]> <azure.devops.v6_0.test_plan.models.[TestSuite]>`
        """

如何检索继续标记以继续解析其他结果?
注意:我还在Azure DevOps Python API的GitHub存储库中创建了一个问题:https://github.com/microsoft/azure-devops-python-api/issues/461

k4emjkb1

k4emjkb11#

我目前对client.Client类做了一个丑陋的补丁,添加了一个属性,在每次发送请求时(使用方法_send)存储一个延续令牌。这是超级丑陋,但至少它的工作

"""Patching ADO Client to retrieve continuation token

Related to question in following issue:
https://github.com/microsoft/azure-devops-python-api/issues/461
"""
import logging
from typing import Optional, cast

from azure.devops import _models
from azure.devops.client import Client
from azure.devops.client_configuration import ClientConfiguration
from msrest import Deserializer, Serializer
from msrest.service_client import ServiceClient

logger = logging.getLogger("azure.devops.client")

# pylint: disable=super-init-not-called

class ClientPatch(Client):
    """Client.
    :param str base_url: Service URL
    :param Authentication creds: Authenticated credentials.
    """

    def __init__(self, base_url=None, creds=None):
        self.config = ClientConfiguration(base_url)
        self.config.credentials = creds
        self._client = ServiceClient(creds, config=self.config)
        _base_client_models = {
            k: v for k, v in _models.__dict__.items() if isinstance(v, type)
        }
        self._base_deserialize = Deserializer(_base_client_models)
        self._base_serialize = Serializer(_base_client_models)
        self._all_host_types_locations = {}
        self._locations = {}
        self._suppress_fedauth_redirect = True
        self._force_msa_pass_through = True
        self.normalized_url = Client._normalize_url(base_url)
        self.continuation_token_last_request: Optional[str] = None

    def _send(
        self,
        http_method,
        location_id,
        version,
        route_values=None,
        query_parameters=None,
        content=None,
        media_type="application/json",
        accept_media_type="application/json",
        additional_headers=None,
    ):
        request = self._create_request_message(
            http_method=http_method,
            location_id=location_id,
            route_values=route_values,
            query_parameters=query_parameters,
        )
        negotiated_version = self._negotiate_request_version(
            self._get_resource_location(self.normalized_url, location_id), version
        )
        negotiated_version = cast(str, negotiated_version)

        if version != negotiated_version:
            logger.info(
                "Negotiated api version from '%s' down to '%s'."
                " This means the client is newer than the server.",
                version,
                negotiated_version,
            )
        else:
            logger.debug("Api version '%s'", negotiated_version)

        # Construct headers
        headers = {
            "Content-Type": media_type + "; charset=utf-8",
            "Accept": accept_media_type + ";api-version=" + negotiated_version,
        }
        if additional_headers is not None:
            for key in additional_headers:
                headers[key] = str(additional_headers[key])
        if self.config.additional_headers is not None:
            for key in self.config.additional_headers:
                headers[key] = self.config.additional_headers[key]
        if self._suppress_fedauth_redirect:
            headers["X-TFS-FedAuthRedirect"] = "Suppress"
        if self._force_msa_pass_through:
            headers["X-VSS-ForceMsaPassThrough"] = "true"
        if (
            Client._session_header_key in Client._session_data
            and Client._session_header_key not in headers
        ):
            headers[Client._session_header_key] = Client._session_data[
                Client._session_header_key
            ]
        response = self._send_request(
            request=request, headers=headers, content=content, media_type=media_type
        )
        if Client._session_header_key in response.headers:
            Client._session_data[Client._session_header_key] = response.headers[
                Client._session_header_key
            ]

        # Patch: Workaround to be able to see the continuation token of the response
        self.continuation_token_last_request = self._get_continuation_token(response)

        return response


def patch_azure_devops_client():
    """Patch the Azure DevOps client to see the continuation token of the response"""
    # pylint: disable=protected-access
    Client.__init__ = ClientPatch.__init__  # type: ignore
    Client._send = ClientPatch._send  # type: ignore

然后,在创建客户端之前,您只需要导入并调用patch_azure_devops_client函数,它将向客户端添加continuation_token_last_request。

相关问题