151 lines
4.2 KiB
Python
Executable File
151 lines
4.2 KiB
Python
Executable File
#!/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
|
|
DRY_RUN = True
|
|
|
|
|
|
# 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'][-1]
|
|
|
|
|
|
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')
|
|
|
|
def img_name(self, svc_name, top):
|
|
return 'enevodocker/enevoleads{}e:{}_{}'.format(
|
|
'f' if 'front' in svc_name else 'b',
|
|
'prod' if self.is_prod else 'stage',
|
|
self.get_build_id(svc_name, top))
|
|
|
|
def get_build_id(self, service, top):
|
|
return self.az.builds(SERVICE_MAPPER[service], top)['id']
|
|
|
|
def get_svc(self, name):
|
|
for svc in self.svcs:
|
|
if svc.name == name:
|
|
return svc
|
|
|
|
def deploy(self, name, top=1):
|
|
img = self.img_name(name, top)
|
|
print('Deploying image {} on service {}'.format(img, name))
|
|
return self.get_svc(name).update(image=img, name=name)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
from getopt import GetoptError, getopt
|
|
from sys import argv
|
|
|
|
def usage():
|
|
print(
|
|
'Usage: {} [option(s)]\n'.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-r{x}\tDeploy x number of builds back' +
|
|
'\n\t-h\tShow this help')
|
|
exit(1)
|
|
|
|
is_prod = False
|
|
svcs = []
|
|
top = 1
|
|
|
|
try:
|
|
opts, args = getopt(argv[1:], 'pfbnhr:')
|
|
except GetoptError as exc:
|
|
print('\nError: {}\n'.format(exc))
|
|
usage()
|
|
|
|
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 == '-r':
|
|
top += int(arg)
|
|
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, top), svc))
|
|
exit(0)
|
|
|
|
for svc in svcs:
|
|
res = dep.deploy(svc, top)
|
|
|
|
if res['Warnings']:
|
|
from json import dumps
|
|
print(dumps(res, indent=2))
|
|
|
|
print('Done! 🤗')
|