ecs_deploy.py

#!/usr/bin/env python3

"""
Usage:
IMAGE=039718039073.dkr.ecr.us-east-1.amazonaws.com/my-service-api:$BITBUCKET_COMMIT
ecs_deploy.py update_taskdef my-service-api --image=$IMAGE
ecs_deploy.py update_service my-service-api --taskdef=my-service-api
ecs_deploy.py run_task my-service-api-migrate-db
"""

import argparse
import os

import boto3

REGION_NAME = os.environ.get("AWS_REGION_NAME", "")
ECS_CLUSTER = os.environ.get("ECS_CLUSTER", "")
client = boto3.client("ecs", region_name=REGION_NAME)


class Command:
    name = None

    def __init__(self, subparsers):
        parser = subparsers.add_parser(self.name)
        self.add_arguments(parser)

    def add_arguments(self, parser):
        raise NotImplementedError()

    def handle(self, args):
        raise NotImplementedError()


class RunTaskCommand(Command):
    name = "run_task"

    def add_arguments(self, parser):
        parser.add_argument("task")

    def handle(self, args):
        client.run_task(
            cluster=ECS_CLUSTER, taskDefinition=args.task, launchType="EC2", count=1
        )


class UpdateTaskdefCommand(Command):
    name = "update_taskdef"

    def add_arguments(self, parser):
        parser.add_argument("taskdef")
        parser.add_argument("--image", dest="image", required=True)

    def handle(self, args):
        taskdef = self.get_task_definition(args.taskdef)

        taskdef["containerDefinitions"][0]["image"] = args.image
        allowed_args = (
            "family",
            "taskRoleArn",
            "executionRoleArn",
            "networkMode",
            "containerDefinitions",
            "volumes",
            "placementConstraints",
            "requiresCompatibilities",
            "cpu",
            "memory",
        )

        kwargs = {k: taskdef[k] for k in allowed_args if k in taskdef}
        response = client.register_task_definition(**kwargs)
        return response["taskDefinition"]["taskDefinitionArn"]

    def get_task_definition(self, task):
        response = client.describe_task_definition(taskDefinition=task)
        return response["taskDefinition"]


class UpdateServiceCommand(Command):
    name = "update_service"

    def add_arguments(self, parser):
        parser.add_argument("service")
        parser.add_argument("--taskdef", dest="taskdef", required=True)

    def handle(self, args):
        response = client.describe_task_definition(taskDefinition=args.taskdef)
        client.update_service(
            cluster=ECS_CLUSTER,
            service=args.service,
            taskDefinition=response["taskDefinition"]["taskDefinitionArn"],
        )


def main():
    command_classes = [RunTaskCommand, UpdateTaskdefCommand, UpdateServiceCommand]

    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers(dest="command")

    commands = {c.name: c(subparsers) for c in command_classes}

    args = parser.parse_args()
    command = commands.get(args.command)

    if command is not None:
        command.handle(args)


if __name__ == "__main__":
    main()