azure-devops-python-samples/rollupChildrenCompletedState.py at main · KaiWalter/azure-devops-python-samples · GitHub

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
from azure.devops.connection import Connection
from azure.devops.credentials import BasicAuthentication
from azure.devops.v6_0.work_item_tracking.models import Wiql, JsonPatchOperation

import argparse
import datetime
import os
import pprint


def main():

    # extract arguments
    parser = argparse.ArgumentParser(
        description='roll up completed state from child work items to parent')
    parser.add_argument("-o", "--org", required=True, dest="url",
                        help="Azure DevOps Organization URL")
    parser.add_argument("-p", "--project", required=True, dest="project",
                        help="Azure DevOps Project")
    parser.add_argument("-t", "--pat", required=True, dest="pat",
                        help="Azure DevOps Personal Access Token")
    parser.add_argument("--parent-type", required=True, dest="parent_type",
                        help="work item parent type to filter for (Bug,Feature,...)")
    parser.add_argument("--child-type", required=True, dest="child_type",
                        help="work item child type to filter for (Task,Product Backlog Item,...)")
    parser.add_argument("--age", required=False, dest="age", default=120, type=int,
                        help="age in days when last change of work item happened")
    parser.add_argument("--update", required=False, action='store_true',
                        dest="update", help="commit update to Azure DevOps")
    args = parser.parse_args()

    # create a connection to the org
    credentials = BasicAuthentication('', args.pat)
    connection = Connection(base_url=args.url, creds=credentials)

    # get a client
    wit_client = connection.clients.get_work_item_tracking_client()

    # determine potential Completed+Removed states for the parent/child work item type
    wi_types = wit_client.get_work_item_types(args.project)
    parent_completed_states = [s.name for s in [
        t for t in wi_types if t.name == args.parent_type][0].states if s.category == 'Completed']
    child_completed_states = [s.name for s in [
        t for t in wi_types if t.name == args.child_type][0].states if s.category == 'Completed']

    # query relations
    wiql = Wiql(
        query=f"""SELECT *
            FROM workitemLinks
            WHERE [Source].[System.TeamProject] = '{args.project}'
            AND [Source].[System.WorkItemType] = '{args.parent_type}'
            AND [Target].[System.WorkItemType] = '{args.child_type}'
            AND [Source].[System.ChangedDate] >= @today - {args.age}
            AND [System.Links.LinkType] = 'Child'
            MODE (MustContain)
        """
    )

    wi_relations = wit_client.query_by_wiql(wiql, top=1000).work_item_relations
    print(f'Results: {len(wi_relations)}')

    # process relations
    if wi_relations:

        parents = [wir.target for wir in wi_relations if not wir.source]

        for parent in parents:

            parent_fields = wit_client.get_work_item(parent.id).fields

            if not parent_fields['System.State'] in parent_completed_states:

                children = [
                    wir.target.id for wir in wi_relations if wir.source and wir.source.id == parent.id]

                completed_children = [child_fields for child_fields in map(lambda id:wit_client.get_work_item(
                    id).fields, children) if child_fields['System.State'] in child_completed_states]

                print(
                    f"{parent_fields['System.WorkItemType']} {parent.id} : {len(completed_children)} of {len(children)} completed")

                operations = []

                if len(completed_children) == len(children):
                        print(f" =>{parent_completed_states[0]}")
                        operations.append(JsonPatchOperation(
                            op='replace', path=f'/fields/System.State', value=parent_completed_states[0]))

                if len(operations) > 0 and args.update:
                    resp = wit_client.update_work_item(
                        document=operations, id=parent.id)
                    print(resp)


if __name__ == '__main__':
    main()