A Python3 script to create and update GitHub repository deployments

deployment.py

from numbers import Number
import os
import requests
import json
from dotenv import load_dotenv

environments = ["dev", "dev2", "qa", "qa2", "stage", "prod"] # update as needed
deployment_states = ["error", "failure", "inactive", "in_progress", "queued", "pending", "success"]

def get_env_var(env_var):
    load_dotenv()
    return os.getenv(env_var)

def get_github_token():
    token = get_env_var("GITHUB_API_TOKEN")
    if (token is None or token == ""):
        raise ValueError("A valid GitHub token is required")
    
    return token
    
def get_github_repo(endpoint = "deployments"):
    owner = get_env_var("GITHUB_REPO_OWNER")
    repo = get_env_var("GITHUB_REPO_NAME")
    if (owner is None or owner == "" or repo is None or repo == ""):
        raise ValueError("A valid GitHub owner and repo is required")
    
    return f"https://api.github.com/repos/{owner}/{repo}/{endpoint}"

def create_deployment(ref = "main", environment = environments[0], log_url = None, required_contexts = None):
    if (ref is None or type(ref) is not str):
        print("[ERROR] A deployment ref is required")
        return False
    if (environment is None or environment not in environments):
        print("[ERROR] A valid environment is required from the following list: " + ", ".join(environments))
        return False
    if (log_url is not None and type(log_url) is not str):
        print("[ERROR] A valid log_url is required")
        return False
    if (required_contexts is not None and type(required_contexts) is not list):
        print("[ERROR] A valid list of required contexts is required")
        return False
    
    print(f"[INFO] Creating deployment for ref: {ref} in environment: {environment}")

    req = requests.post(get_github_repo("deployments"), headers = {
        "Accept": "application/json",
        "Content-Type": "application/json",
        "Authorization": f"Bearer {get_github_token()}",
        "X-GitHub-Api-Version": "2022-11-28"
    }, data=json.dumps({
        "ref": ref,
        "environment": environment,
        "description": f"Deployment to {environment}",
        "auto_merge": False,
        "required_contexts": required_contexts
    }))

    if (req.status_code != 201):
        print(f"[ERROR] Failed to create deployment: {req.text}")
        return False
    
    try:
        deployment = req.json()
        return deployment
    except:
        print(f"[ERROR] Failed to parse deployment response: {req.text}")
        return False

def update_deployment(id, state = "success", environment_url = ""):
    if (id is None or not int(id)):
        print("[ERROR] A deployment id is required")
        return False
    if (state is None or state not in deployment_states):
        print("[ERROR] A valid state is required from the following list: " + ", ".join(deployment_states))
        return False
    if (environment_url is not None and type(environment_url) is not str):
        print("[ERROR] A valid environment_url is required")
        return False
    
    print(f"[INFO] Updating deployment with id: {id}")

    req = requests.post(get_github_repo(f"deployments/{id}/statuses"), headers = {
        "Accept": "application/json",
        "Content-Type": "application/json",
        "Authorization": f"Bearer {get_github_token()}",
        "X-GitHub-Api-Version": "2022-11-28"
    }, data=json.dumps({
        "state": state,
        "description": f"Deployment {state}",
        "environment_url": environment_url
    }))

    if (req.status_code != 201):
        print(f"[ERROR] Failed to update deployment: {req.text}")
        return False
    
    try:
        deployment = req.json()
        return deployment
    except:
        print(f"[ERROR] Failed to parse deployment response: {req.text}")
        return False

def handle_create_deployment():
        ref = input("Enter the ref for the deployment: ")
        environment = input(f"Enter the environment for the deployment ({', '.join(environments)}): ")
        log_url = input("Enter the log url for the deployment: ")
        result = create_deployment(ref, environment, log_url, [])

        if (result):
            print(f"[INFO] Deployment created with id: {result['id']}")
        else:
            retry = input("Would you like to retry? (y/n): ")
            if (retry == "y"):
                handle_create_deployment()

        return result

def handle_update_deployment(previous_id = None):
    id = input(f"Enter the id of the deployment (Prev: {previous_id if previous_id is not None else 'N/A'}): ")
    state = input(f"Enter the state for the deployment ({', '.join(deployment_states)}): ")
    environment_url = input("Enter the environment url for the deployment: ")
    result = update_deployment(id, state, environment_url)

    if (result):
        print(f"[INFO] Deployment updated with id: {result['id']}")
    else:
        retry = input("Would you like to retry? (y/n): ")
        if (retry == "y"):
            handle_update_deployment()

    return result
        
def main():
    while True:
        previous_id = None
        action = input("Choose an action:\n\t1. Create Deployment\n\t2. Update Deployment\n\t3. Exit\n: ")

        if (action == "1"):
            deployment = handle_create_deployment()
            if (deployment and "id" in deployment):
                previous_id = deployment["id"]
        elif (action == "2"):
            handle_update_deployment(previous_id)
        elif (action == "3"):
            break

if __name__ == "__main__":
    main()