File: ref.py

package info (click to toggle)
azure-devops-cli-extension 1.0.2-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 20,384 kB
  • sloc: python: 160,782; xml: 198; makefile: 56; sh: 51
file content (125 lines) | stat: -rw-r--r-- 5,773 bytes parent folder | download | duplicates (3)
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
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------

from knack.log import get_logger
from knack.util import CLIError
from azext_devops.devops_sdk.v5_0.git.models import GitRefUpdate
from azext_devops.dev.common.git import resolve_git_refs
from azext_devops.dev.common.services import (get_git_client,
                                              resolve_instance_project_and_repo)

logger = get_logger(__name__)


# pylint: disable=redefined-builtin
def list_refs(filter=None, repository=None, organization=None, project=None, detect=None):
    """List the references.
    :param str filter: A filter to apply to the refs (starts with). Example: head or heads/ for the branches.
    :param str repository: Name or ID of the repository.
    :param str project: Name or ID of the project.
    :param str detect: Automatically detect organization and project. Default is "on".
    """
    organization, project, repository = resolve_instance_project_and_repo(
        detect=detect,
        organization=organization,
        project=project,
        repo=repository)
    client = get_git_client(organization)
    return client.get_refs(repository_id=repository,
                           project=project,
                           filter=filter)


def create_ref(name, object_id, repository=None, organization=None, project=None, detect=None):
    """Create a reference.
    :param str name: Name of the reference to create (example: heads/my_branch or tags/my_tag).
    :param str object_id: Id of the object to create the reference from.
    :param str repository: Name or ID of the repository.
    :param str project: Name or ID of the project.
    :param str detect: Automatically detect organization and project. Default is "on".
    """
    organization, project, repository = resolve_instance_project_and_repo(
        detect=detect,
        organization=organization,
        project=project,
        repo=repository)
    client = get_git_client(organization)
    # by default, the create method does not support setting the is_locked value
    # to True.
    ref_update = GitRefUpdate(is_locked=False,
                              name=resolve_git_refs(name),
                              new_object_id=object_id,
                              old_object_id='0000000000000000000000000000000000000000')
    response = client.update_refs(ref_updates=[ref_update],
                                  repository_id=repository,
                                  project=project)[0]
    if response.success is False:
        raise CLIError(response.custom_message)
    return response


def delete_ref(name, object_id=None, repository=None, organization=None, project=None, detect=None):
    """Delete a reference.
    :param str name: Name of the reference to delete (example: heads/my_branch).
    :param str object_id: Id of the reference to delete.
    :param str repository: Name or ID of the repository.
    :param str project: Name or ID of the project.
    :param str detect: Automatically detect organization and project. Default is "on".
    """
    organization, project, repository = resolve_instance_project_and_repo(
        detect=detect,
        organization=organization,
        project=project,
        repo=repository)
    client = get_git_client(organization)

    if object_id is None:
        ref = client.get_refs(repository_id=repository, project=project, filter=name)
        if not ref or len(ref) != 1:
            logger.error('ref not found')
            raise CLIError("Failed to find object_id for ref " + name + ". Please provide object_id.")

        object_id = ref[0].object_id

    ref_update = GitRefUpdate(name=resolve_git_refs(name),
                              new_object_id='0000000000000000000000000000000000000000',
                              old_object_id=object_id)
    return client.update_refs(ref_updates=[ref_update],
                              repository_id=repository,
                              project=project)[0]


def lock_ref(name, repository=None, organization=None, project=None, detect=None):
    """Lock a reference.
    :param str name: Name of the reference to update (example: heads/my_branch).
    :param str repository: Name or ID of the repository.
    :param str project: Name or ID of the project.
    :param str detect: Automatically detect organization and project. Default is "on".
    """
    return _update_ref(name, True, repository, organization, project, detect)


def unlock_ref(name, repository=None, organization=None, project=None, detect=None):
    """Unlock a reference.
    :param str name: Name of the reference to update (example: heads/my_branch).
    :param str repository: Name or ID of the repository.
    :param str project: Name or ID of the project.
    :param str detect: Automatically detect organization and project. Default is "on".
    """
    return _update_ref(name, False, repository, organization, project, detect)


def _update_ref(name, locked, repository, organization, project, detect):
    organization, project, repository = resolve_instance_project_and_repo(
        detect=detect,
        organization=organization,
        project=project,
        repo=repository)
    client = get_git_client(organization)
    ref_update = GitRefUpdate(is_locked=locked)
    return client.update_ref(new_ref_info=ref_update,
                             repository_id=repository,
                             filter=name,
                             project=project)