AzDevopsAPIWrapper/domain.py at master · Lazymindz/AzDevopsAPIWrapper · 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
import requests
from requests.auth import HTTPProxyAuth
import json
import os, os.path
import logging
from logging import NullHandler
from logging.config import fileConfig

# Set default logging handler to avoid "No handler found" warnings.
logging.getLogger(__name__).addHandler(NullHandler())


class AzureDevops(object):
    """DevOps Build Pipeline Class"""

    def __init__(self, org_name, project_name):
        self.org_name = org_name
        self.project_name = project_name
        self.auth = ''
        self.proxy = {
            'http'  : '',
            'https' : ''
            }
        self.proxyauth = None

    def setauth(self, auth):
        self.auth = auth
        return None

    def setproxy(self, proxy, proxyauth):
        self.proxy['http'] = proxy
        self.proxy['https'] = proxy
        self.proxyauth = proxyauth
        return None

    def getbuilds(self):
        requesturl = 'https://dev.azure.com/{organization}/{project}/_apis/build/builds?api-version=5.0'.format(organization=self.org_name, project=self.project_name)
        headers = {
            'Authorization': self.auth,
            }
        response = requests.request("GET", url=requesturl, headers=headers, proxies=self.proxy, auth=self.proxyauth)
        return response.content

    def getbuildids(self):
        builds = json.loads(self.getbuilds())
        buildids = [build['id'] for build in builds['value']]
        logging.debug('Build Ids extracted..')
        return buildids

    def getbuildartifacts(self, buildid):
        requesturl = 'https://dev.azure.com/{organization}/{project}/_apis/build/builds/{buildId}/artifacts?api-version=5.0'.format(
            organization=self.org_name,
            project=self.project_name,
            buildId=buildid
        )
        headers = {
            'Authorization': self.auth,
            }
        response = requests.request("GET", url=requesturl, headers=headers, proxies=self.proxy, auth=self.proxyauth)
        return json.loads(response.content)

    def downloadartifact(self, buildid, artifact):
        requesturl = artifact['resource']['downloadUrl']
        filename = '{name}.zip'.format(name = artifact['name'])
        filepath = './BuildArtifacts/{buildid}/{filename}'.format(filename=filename, buildid=buildid)
        headers = {
            'Authorization': self.auth,
            }
        response = requests.request("GET", url=requesturl, headers=headers, proxies=self.proxy, auth=self.proxyauth)

        if not os.path.exists("./BuildArtifacts/{buildid}".format(buildid=buildid)):
            os.makedirs("./BuildArtifacts/{buildid}".format(buildid=buildid))

        with open(filepath, 'wb') as f:
            f.write(response.content)

        return None