This script will take a Humanitec application as a blueprint and create a new Humanitec application from it based on the last deployment set. Learn more about Humanitec at www.humanitec.com

copy-hum-app-blueprint.py

"""
This script will take a Humanitec application as a blueprint
and create a new Humanitec application from it based on the
last deployment set in each environment.
Also the pipelines of the application will be copied.

It's required to have your HUMANITEC_TOKEN set as an environment variable

The script takes 3 parameters
1. your Humanitec organization id
2. your Humanitec application id, which will be used as the blueprint
3. a new application id, which will be created as a copy of the blueprint
"""

import requests
import os
import sys

try:
    humanitec_token = os.environ["HUMANITEC_TOKEN"]
except Exception as e:
    print(f"Error: Could not read {e} from environment.")
    print(f"Please export {e} as environment variable.")
    sys.exit()

if len(sys.argv) != 4:
    print(
        f" Usage: python3 {sys.argv[0]} humanitec-org-id humanitec-app-id new-humanitec-app-id"
    )
    print(f" Example: python3 {sys.argv[0]} my-org my-blueprint-app new-app")
    sys.exit(0)

org = sys.argv[1]
app = sys.argv[2]
new_app = sys.argv[3]

humanitec_url = "api.humanitec.io"

headers = {
    "Authorization": f"Bearer {humanitec_token}",
    "Content-Type": "application/json",
}


def get_envs():
    """
    Get all environment of your application blueprint.
    """
    envs = []
    url = f"https://{humanitec_url}/orgs/{org}/apps/{app}"
    response = requests.request("GET", url, headers=headers)
    if response.status_code == 200:
        for env in response.json()["envs"]:
            envs.append(env["id"])
    else:
        sys.exit(
            f"Unable to get application. GET {url} returned status code {response.status_code}."
        )
    return envs


def get_depl_set(env):
    """
    Get the latest deployment set of an environment.
    """
    url = f"https://{humanitec_url}/orgs/{org}/apps/{app}/envs/{env}"
    response = requests.request("GET", url, headers=headers)
    if response.status_code == 200 and response.json()["from_deploy"]:
        set_id = response.json()["from_deploy"]["set_id"]
    else:
        sys.exit(
            f"Unable to get deployment set. GET {url} returned status code {response.status_code}."
        )
    return set_id


def get_pipelines():
    """
    Get all pipeline ids in your blueprint application. (Ignores default pipelines)
    """
    pipeline_ids = []
    url = f"https://{humanitec_url}/orgs/{org}/apps/{app}/pipelines"
    response = requests.request("GET", url, headers=headers)
    if response.status_code == 200:
        for pipeline in response.json():
            if pipeline["id"] != "default":
                pipeline_ids.append(pipeline["id"])
    else:
        sys.exit(
            f"Unable to get pipelines. GET {url} returned status code {response.status_code}."
        )
    return pipeline_ids


def get_pipeline_schema(pipeline_id):
    """
    Get the pipeline schema of a pipeline id.
    """
    url = (
        f"https://{humanitec_url}/orgs/{org}/apps/{app}/pipelines/{pipeline_id}/schema"
    )
    response = requests.request("GET", url, headers=headers)
    if response.status_code == 200:
        pipeline_schema = response.content
        pipeline_schema = pipeline_schema.decode("utf-8")
    else:
        sys.exit(
            f"Unable to get pipeline schema. GET {url} returned status code {response.status_code}."
        )
    return pipeline_schema


def get_set_content(set_id):
    """
    Get the content of a deployment set's modules section.
    """
    url = f"https://{humanitec_url}/orgs/{org}/apps/{app}/sets/{set_id}"
    response = requests.request("GET", url, headers=headers)
    if response.status_code == 200:
        modules = response.json()["modules"]
    else:
        sys.exit(
            f"Unable to get application deployment set. GET {url} returned status code {response.status_code}."
        )
    return modules


def create_env(app, env, from_deploy_id):
    """
    Create new application environments.
    """
    url = f"https://{humanitec_url}/orgs/{org}/apps/{app}/envs"
    payload = {
        "id": f"{env}",
        "name": f"{env}",
        "type": "development",
        "from_deploy_id": f"{from_deploy_id}",
    }
    response = requests.request("POST", url, headers=headers, json=payload)
    if response.status_code == 201:
        print(f"The environment {env} has been created.")
    else:
        sys.exit(
            f"Unable to create environment. POST {url} returned status code {response.status_code}."
        )


def create_pipeline(app, pipeline_schema):
    """
    Create new pipeline.
    """
    url = f"https://{humanitec_url}/orgs/{org}/apps/{app}/pipelines"
    headers = {
        "Authorization": f"Bearer {humanitec_token}",
        "Content-Type": "application/x-yaml",
    }
    payload = pipeline_schema
    response = requests.request("POST", url, headers=headers, data=payload)
    if response.status_code == 201:
        print(f"The pipeline {response.json()['name']} has been created.")
    else:
        sys.exit(
            f"Unable to create pipeline. POST {url} returned status code {response.status_code}."
        )


# Create application
url = f"https://{humanitec_url}/orgs/{org}/apps"
payload = {"id": f"{new_app}", "name": f"{new_app}"}
response = requests.request("POST", url, headers=headers, json=payload)
if response.status_code == 201:
    print(f"The application {new_app} has been created.")
    app_id = response.json()["id"]
else:
    sys.exit(
        f"Unable to create application. POST {url} returned status code {response.status_code}."
    )

# Create environments (ignore development)
envs = get_envs()
for env in envs:
    if env != "development":
        create_env(new_app, env, deploy_id)
    set_id = get_depl_set(env)
    modules = get_set_content(set_id)

    # Create delta in new application
    url = f"https://{humanitec_url}/orgs/{org}/apps/{new_app}/deltas"
    payload = {"modules": {"add": modules}}
    response = requests.request("POST", url, headers=headers, json=payload)
    if response.status_code == 200:
        delta_id = response.json()
        print(delta_id)
    else:
        sys.exit(
            f"Unable to create delta. POST {url} returned status code {response.status_code}."
        )

    # Trigger the deployment of the deployment delta
    url = f"https://{humanitec_url}/orgs/{org}/apps/{new_app}/envs/{env}/deploys"
    payload = {"delta_id": f"{delta_id}", "comment": "init deploy"}
    response = requests.request("POST", url, headers=headers, json=payload)
    if response.status_code == 201:
        print(f"The deployment for application {app_id} has been triggered.")
        deploy_id = response.json()["id"]
    else:
        sys.exit(
            f"Unable to trigger deployment. POST {url} returned status code {response.status_code}."
        )

# Create pipelines
pipeline_ids = get_pipelines()
for pipeline_id in pipeline_ids:
    pipeline_schema = get_pipeline_schema(pipeline_id)
    create_pipeline(new_app, pipeline_schema)