DeployTests/deploy/portainer/deploy.py

#!/usr/bin/env python3

import os
import sys
import argparse
import requests
import json
import uuid
from dotenv import load_dotenv
from string import Template
from pathlib import Path
from urllib.parse import urlparse

load_dotenv()

required_env_vars = {
    'PORTAINER': 'The portainer instance to deploy to',
    'PORTAINER_API_KEY': 'API-Key to access portainer instance',
    'PORTAINER_EP': 'Portainer Environment EndPoint to deploy to',
    'GITEA_API_KEY': 'API-Key to access Gitea instance',
    'DEPLOY_REPO_URL': 'The repository URL to deploy',
    'DEPLOY_REF': 'The git ref to deploy',
    'DEPLOY_HOST': 'The host name under which the deployment will be reachable',
    'DEPLOY_NAME': 'Custom name to use as the deployment name',
}

# Try getting all arguments from (in order): 1 command line, 2 .env file, 3 Environment
parser = argparse.ArgumentParser(description='Deploys a docker compose application to portainer.')

for var, usage in required_env_vars.items():
    parser.add_argument(f'--{var}', default=os.getenv(var, None), help=usage)
args = parser.parse_args()

# Check if all were parsed
not_parsed = []
for var, usage in required_env_vars.items():
    if not getattr(args, var):
        not_parsed.append(var)
    else:
        print(f'--{var}: {getattr(args, var)}')

if not_parsed:
    print(f"Error: The following required environment variables were not provided: {', '.join(not_parsed)}")
    parser.print_help()
    sys.exit(1)

# Deploy variables to substitute in portainer deploy template
deploy_variables = {key: getattr(args, key) for key in 
    ['DEPLOY_REPO_URL', 'DEPLOY_HOST', 'DEPLOY_NAME', 'DEPLOY_REF']
}
deploy_variables['DEPLOY_WEBHOOK'] = str(uuid.uuid4())

portainer=args.PORTAINER
portainer_api_key=args.PORTAINER_API_KEY
portainer_ep=args.PORTAINER_EP
gitea_api_key=args.GITEA_API_KEY
deploy_repo=deploy_variables['DEPLOY_REPO_URL']
deploy_webhook=deploy_variables['DEPLOY_WEBHOOK']
deploy_ref=deploy_variables['DEPLOY_REF']

### Find CICD-runner portainer environment endpointId ###
portainer_headers = {
    'Content-Type': 'application/json',
    'X-API-Key': portainer_api_key,
}
endpoint_url = f'{portainer}/api/endpoints'
json_endpoints = None
try:
    response = requests.get(endpoint_url, headers=portainer_headers)
    response.raise_for_status()  # Raise HTTPError for bad requests
    json_endpoints = response.json()
    
except requests.exceptions.RequestException as err:
    raise Exception(f'Could not retrieve portainer endpoints: {err}')

endpoint_id = None
for endpoint in json_endpoints:
    if endpoint["Name"] == portainer_ep:
        endpoint_id = endpoint["Id"]
        break
if endpoint_id is None:
    raise Exception(f'Portainer endpoint \'{portainer_ep}\' not found.')
else:
    print(f'Found portainer endpoint \'{portainer_ep}\' has id: \'{endpoint_id}\'.')


### Template substitution for the portainer stack deployment ###
portainer_deploy_payload = {
  "additionalFiles": [
    "deploy/portainer/portainer_deploy.docker-compose.yml"
  ],
  "autoUpdate": {
    "webhook": f"{deploy_variables['DEPLOY_WEBHOOK']}"
  },
  "composeFile": "docker-compose.yml",
  "env": [
    {
      "name": "HOST",
      "value": f"{deploy_variables['DEPLOY_HOST']}"
    },
    {
      "name": "COMPOSE_PROJECT_NAME",
      "value": f"{deploy_variables['DEPLOY_NAME']}"
    }
  ],
  "fromAppTemplate": False,
  "name": f"{deploy_variables['DEPLOY_NAME']}",
  "repositoryAuthentication": True,
  "repositoryUsername": "cicd",
  "repositoryPassword": "gJ6@$7ZjWGyV4%i", 
  "repositoryReferenceName": f"{deploy_variables['DEPLOY_REF']}",
  "repositoryURL": f"{deploy_variables['DEPLOY_REPO_URL']}",
  "tlsskipVerify": False
}

### Deploy to portainer ###
deploy_url = f'{portainer}/api/stacks/create/standalone/repository?endpointId={endpoint_id}'
try:
    response = requests.post(deploy_url, headers=portainer_headers, json=portainer_deploy_payload)
    response.raise_for_status()  # Raise HTTPError for bad requests
    deploy_response = response.json()

except requests.exceptions.RequestException as err:
    raise Exception(f'Could not deploy portainer stack: {err}')

### Add Webhook to Gitea ###
repo_url = urlparse(deploy_repo)
gitea = f"{repo_url.scheme}://{repo_url.netloc}"
repo_path = repo_url.path
repo_parts = repo_path.strip('/').split('/')
owner = repo_parts[0]
repo = repo_parts[1]

ref_parts = deploy_ref.strip('/').split('/')
branch = ref_parts[-1]

webhook_payload = {
    "type": "gitea",
    "branch_filter": f"{branch}",
    "config": {
        "url": f"{portainer}/api/stacks/webhooks/{deploy_webhook}",
        "content_type": "json"
    },
    "events": ["push"],  # You can specify other events as needed
    "active": True
}

webhook_url = f'{gitea}/api/v1/repos/{owner}/{repo}/hooks'
webhook_headers = {
    "Authorization": f"token {gitea_api_key}"
}
try:
    response = requests.post(webhook_url, headers=webhook_headers, json=webhook_payload)
    response.raise_for_status()  # Raise HTTPError for bad requests
    webhook_response = response.json()

except requests.exceptions.RequestException as err:
    raise Exception(f'Could not add webhook to Gitea: {err}')
print(f'Successfully deployed project')