Trigger Code Deploy

codedeploy.py

import boto3
import sys
import time
from os import getenv

region_name = getenv("PROJECT_AWS_REGION")
account_id = getenv("PROJECT_AWS_ACCOUNTID")
assume_role_name = getenv("PROJECT_AWS_IAM_ROLENAME")
aws_access_key_id = getenv("PROJECT_AWS_ACCOUNT_ACCESSKEY")
aws_secret_access_key = getenv("PROJECT_AWS_ACCOUNT_SECRETKEY")
code_deploy_app_name = getenv("PROJECT_AWS_CODEDEPLOY_APPLICATIONNAME")
code_deploy_dg_name = getenv("PROJECT_AWS_CODEDEPLOY_DEPLOYMENTGROUPNAME")

package_id = getenv("PROJECT_RELEASE_PACKAGEID")
package_version = getenv("PROJECT_RELEASE_PACKAGEID_PACKAGEVERSION")
artifact_bucket = getenv("PROJECT_ARTIFACT_BUCKET")
artifact_revision = {
    "revisionType": "S3",
    "s3Location": {
        "bucket": artifact_bucket,
        "key": f"{package_id}.{package_version}.zip",
        "bundleType": "zip",
    },
}


def authenticate(
    account_id,
    assume_role_name,
    aws_access_key_id,
    aws_secret_access_key,
    region_name="us-east-1",
    session_name="aws-app",
):
    sts_client = boto3.client(
        "sts",
        aws_access_key_id=aws_access_key_id,
        aws_secret_access_key=aws_secret_access_key,
        region_name=region_name,
    )

    assumed_role = sts_client.assume_role(
        RoleArn=f"arn:aws:iam::{account_id}:role/{assume_role_name}",
        RoleSessionName=session_name,
    )
    return assumed_role["Credentials"]


# Get credentials
credentials = authenticate(
    account_id=account_id,
    assume_role_name=assume_role_name,
    aws_access_key_id=aws_access_key_id,
    aws_secret_access_key=aws_secret_access_key,
    region_name=region_name,
)

# Get codedeploy client
codedeploy_client = boto3.client(
    "codedeploy",
    aws_access_key_id=credentials["AccessKeyId"],
    aws_secret_access_key=credentials["SecretAccessKey"],
    aws_session_token=credentials["SessionToken"],
    region_name=region_name,
)

# Create deployment
print("Creating deployment")
response = codedeploy_client.create_deployment(
    applicationName=code_deploy_app_name,
    deploymentGroupName=code_deploy_dg_name,
    revision=artifact_revision,
)

# Check for deployment status
deployment_id = response.get("deploymentId")
deployment_threshold = 1800  # Max 30 mins threshold
elapsed_secs = 0
check_period = 5  # Check every 5 sec interval
deployment_status = None

print(f"Deployment created with id: {deployment_id}")

# Check until the deployment is in expected status
while elapsed_secs <= deployment_threshold:
    elapsed_secs += check_period

    deployment_status = codedeploy_client.get_deployment(deploymentId=deployment_id)[
        "deploymentInfo"
    ]["status"]

    if deployment_status in ["Succeeded", "Failed", "Stopped"]:
        print(f"Deployment finisehd with status: {deployment_status}")
        break

    print(f"Code Deployment in Progress, status: {deployment_status}")
    time.sleep(check_period)
else:
    print("Status", "Failed")
    raise Exception(
        "Deployment failed, Reason: Deployment time exceeded than threshold"
    )

# Set output variable for next step
print("Status", "deployment_status")

if deployment_status == "Succeeded":
    sys.exit(0)

# Deployment has failed otherwise
sys.exit(5)