coordinating-ecs-task-v1.py

class EcsTask:
    def __init__(self, cluster, task, securityGroup, vpcSubnet, queue, environment):
        self.cluster = cluster
        self.task = task
        self.securityGroup = securityGroup
        self.vpcSubnet = vpcSubnet
        self.queue = queue
        self.environment = environment

    def input(self):
        return 'dict'

    def output(self):
        return 'dict'

    def bind(self, prev, next, metrics, metadata):
        self.prev = prev
        self.next = next
        self.metrics = metrics
        self.ecs = client('ecs')
        self.prev.subscribe(self.changed)

    def changed(self):
        item = self.queue.get_nowait()
        self.metrics.log(f'acquired {item["Host"]}')

        response = self.ecs.run_task(
            cluster=self.cluster,
            taskDefinition=self.task,
            launchType='FARGATE',
            platformVersion='1.4.0',
            networkConfiguration={
                'awsvpcConfiguration': {
                    'assignPublicIp': 'ENABLED',
                    'securityGroups': [self.securityGroup],
                    'subnets': [self.vpcSubnet]
                }
            },
            overrides={
                'containerOverrides': [{
                    'name': self.task.replace('/', ':').split(':')[-2],
                    'environment': self.environment(item)
                }]
            }
        )

        self.metrics.log(f'waiting {response["tasks"][0]["taskArn"]}')
        self.ecs.get_waiter('tasks_stopped').wait(
            cluster=self.cluster,
            tasks=[response['tasks'][0]['taskArn']],
            WaiterConfig={
                'Delay': 6,
                'MaxAttempts': 1200
            }
        )

        self.queue.put(item)
        self.metrics.log(f'released {item["Host"]}')

    def flush(self):
        pass