DeployTests/deploy/portainer/undeploy.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('.env.deploy')

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_BRANCH': 'The branch to deploy'
}

# 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)

portainer_headers = {
    'Content-Type': 'application/json',
    'X-API-Key': args.PORTAINER_API_KEY
}
endpoint_url = f'{args.PORTAINER}/api/endpoints'
try:
    endpoint_response = requests.get(endpoint_url, headers=portainer_headers)
    endpoint_response.raise_for_status()  # Raise HTTPError for bad requests
    json_endpoints = endpoint_response.json()

except requests.exceptions.RequestException as err:
    raise Exception(f'Could not retrieve portainer endpoints: {err}.\n\n Response: {endpoint_response.content}')
 
endpoint_id = None
for endpoint in json_endpoints:
    if endpoint["Name"] == args.PORTAINER_EP:
        endpoint_id = endpoint["Id"]
        break
if endpoint_id is None:
    raise Exception(f'Portainer endpoint \'{args.PORTAINER_EP}\' not found.')
else:
    print(f'Found portainer endpoint \'{args.PORTAINER_EP}\' has id: \'{endpoint_id}\'.')

# ?filters=\{'EndpointId':'{endpoint_id}'\}
stacks_url = f"{args.PORTAINER}/api/stacks"
try:
    stacks_response = requests.get(stacks_url, headers=portainer_headers)
    stacks_response.raise_for_status()  # Raise HTTPError for bad requests
    json_stacks = stacks_response.json()

except requests.exceptions.RequestException as err:
    raise Exception(f'Could not retrieve portainer stacks: {err}.\n\n Response: {stacks_response.content}')

stack_id = None
stack_webhook = None
for stack in json_stacks:
    if (
        stack['GitConfig'] is not None
        and stack['GitConfig']['URL'] == args.DEPLOY_REPO_URL
        and stack['GitConfig']['ReferenceName'] == f"refs/heads/{args.DEPLOY_BRANCH}"
        ):
        stack_id = stack["Id"]
        stack_webhook = stack["AutoUpdate"]["Webhook"]
        break;

if stack_id is None or stack_webhook is None:
    raise Exception(f"Portainer stack with url:'{args.DEPLOY_REPO_URL}' and branch:'{args.DEPLOY_BRANCH}' not found.")
else:
    print(f"Found portainer stack to remove with url:'{args.DEPLOY_REPO_URL}' and branch:'{args.DEPLOY_BRANCH}', stack has id: '{stack_id}' and webhook: '{stack_webhook}'")

### Find correct webhook in gitea
repo_url = urlparse(args.DEPLOY_REPO_URL)
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]

gitea_headers = {
    "Authorization": f"token {args.GITEA_API_KEY}"
}

webhook_url = f'{gitea}/api/v1/repos/{repo_path}/hooks'
try:
    #TODO: Webhooks are returned paginated, this only checks first page
    get_webhooks_response = requests.get(webhook_url, headers=gitea_headers)
    get_webhooks_response.raise_for_status()  # Raise HTTPError for bad requests
    json_webhooks = get_webhooks_response.json()

except requests.exceptions.RequestException as err:
    raise Exception(f'Could not get webhooks from Gitea: {err}.\n\n Response:{get_webhooks_response.content}')

webhook_id = None
for webhook in json_webhooks:
    if webhook["config"]["url"] == f"{args.PORTAINER}/api/stacks/webhooks/{stack_webhook}":
        webhook_id = webhook["id"]
        break
        
if webhook_id is None:
    raise Exception(f"Gitea webhook pointing to Portainer webhook '{stack_webhook}' not found.")
else:
    print(f"Found Gitea webhook pointing to Portainer webhook '{stack_webhook}\' has id: '{webhook_id}'.")

### Remove Webhook from Gitea ###
remove_webhook_url = f"{gitea}/api/v1/repos/{repo_path}/hooks/{webhook_id}"
try:
    del_webhooks_response = requests.delete(remove_webhook_url, headers=gitea_headers)
    del_webhooks_response.raise_for_status()  # Raise HTTPError for bad requests

except requests.exceptions.RequestException as err:
    raise Exception(f"Could not delete webhook '{webhook_id}' from Gitea: {err}.\n\n Response: {del_webhooks_response.content}")

## Remove Stack from Portainer ###
remove_stack_url = f"{args.PORTAINER}/api/stacks/{stack_id}?endpointId={endpoint_id}"
try:
    del_stack_response = requests.delete(remove_stack_url, headers=portainer_headers)
    del_stack_response.raise_for_status()  # Raise HTTPError for bad requests

except requests.exceptions.RequestException as err:
    raise Exception(f"Could not delete stack '{stack_id}' from Portainer: {err}.\n\n Response: {del_stack_response.content}")
    
print(f'Successfully undeployed project')