batman

bin/deploy.py

@@ -0,0 +1,139 @@
+#!/usr/bin/env python3
+from docker import from_env as docker_from_env, DockerClient
+from docker.models.services import Service
+from requests import Session
+from base64 import b64encode
+from typing import List
+from os import environ
+
+
+DRY_RUN = False
+
+
+# Map of Docker service names to Azure DevOps CI pipeline definition IDs
+SERVICE_MAPPER = {
+    'byndle_demofrontend': 18,
+    'byndle_demobackend': 16,
+    'byndle_frontend': 17,
+    'byndle_backend': 14,
+    'byndle_nextcom': 14,
+}
+
+
+class DevOpsAPI: # https://learn.microsoft.com/en-us/rest/api/azure/devops
+    "Limited Azure DevOps API client"
+    base: str = 'https://dev.azure.com/enevo'
+    proj: str = 'Byndle'
+    ver: str = '7.1'
+    ses: Session
+
+    def __init__(self, token=environ.get('AZDO_PAT')):
+        if token is None:
+            raise Exception(
+                    'env var AZDO_PAT needs to be set to a valid Azure DevOps access token!')
+
+        self.ses = Session()
+        self.ses.params['api-version'] = self.ver
+        self.set_token(token)
+
+    def set_token(self, token):
+        self.ses.headers['Authorization'] = 'Basic ' + b64encode(
+            bytes(':' + token, 'UTF-8')).decode()
+
+    def builds(self, definition, top=1):
+        json = self.ses.get('{}/{}/_apis/build/builds?definitions={}&$top={}'.format(
+            self.base, self.proj, definition, top
+        )).json()
+        return json['value'][0] if top == 1 else json
+
+
+class Deployer: # https://docker-py.readthedocs.io/en/stable/index.html
+    "Uses DevOpsAPI to fetch build IDs needed to deploy Docker swarm"
+    svcs: List[Service]
+    dock: DockerClient
+    az: DevOpsAPI
+    is_prod: bool
+
+    def __init__(self, is_prod = False):
+        if environ.get('DOCKER_HOST') is None:
+            raise Exception(
+                    'env var DOCKER_HOST needs to be set to deploy Docker swarm!')
+
+        self.is_prod = is_prod
+        self.dock = docker_from_env()
+        self.az = DevOpsAPI()
+        self.svcs = self.dock.services.list()
+
+        self.svc_name = lambda is_front: 'byndle_{}{}end'.format(
+            '' if self.is_prod else 'demo',
+            'front' if is_front else 'back')
+        
+        self.img_name = lambda svc_name: 'enevodocker/enevoleads{}e:{}_{}'.format(
+            'f' if 'front' in svc_name else 'b',
+            'prod' if self.is_prod else 'stage',
+            self.last_build_id(svc_name))
+
+    def last_build_id(self, service):
+        return self.az.builds(SERVICE_MAPPER[service])['id']
+
+    def get_svc(self, name):
+        for svc in self.svcs:
+            if svc.name == name:
+                return svc
+
+    def deploy(self, name):
+        img = self.img_name(name)
+        print('Deploying image {} on service {}'.format(img, name))
+        return self.get_svc(name).update(image=img, name=name)
+
+
+if __name__ == '__main__':
+    from getopt import getopt
+    from sys import argv
+
+    def usage():
+        print(
+            'Usage: {} [option(s)]'.format(argv[0]) +
+            '\n\t-p\tUse production environment' +
+            '\n\t-f\tDeploy frontend service' +
+            '\n\t-b\tDeploy backend service' +
+            '\n\t-n\tDeploy nextcom service' +
+            '\n\t-h\tShow this help')
+        exit(1)
+
+    is_prod = False
+    svcs = []
+
+    opts, args = getopt(argv[1:], 'pfbnh')
+    if '-p' in [opt for opt, _arg in opts]:
+        is_prod = True
+
+    dep = Deployer(is_prod)
+
+    for opt, arg in opts:
+        if opt == '-f':
+            svcs.append(dep.svc_name(True))
+        elif opt == '-b':
+            svcs.append(dep.svc_name(False))
+        elif opt == '-n':
+            svcs.append('byndle_nextcom')
+        elif opt == '-h':
+            usage()
+
+    if len(svcs) == 0:
+        usage()
+
+    if DRY_RUN:
+        for svc in svcs:
+            print('Would deploy image {} on service {}'.format(
+                dep.img_name(svc), svc))
+        exit(0)
+
+    for svc in svcs:
+        res = dep.deploy(svc)
+
+        if res['Warnings']:
+            from json import dumps
+            print(dumps(res, indent=2))
+
+    print('Done! 🤗')