deploy_new_image.py

import boto3
import uuid

def create_session(role_arn):
    client = boto3.client('sts')
    
    response = client.assume_role(
        RoleArn=role_arn,
        RoleSessionName=str(uuid.uuid1())
    )
    
    return boto3.session.Session(
        aws_access_key_id=response['Credentials']['AccessKeyId'],
        aws_secret_access_key=response['Credentials']['SecretAccessKey'],
        aws_session_token=response['Credentials']['SessionToken']
    )

def _get_attributes(target_dict, attributes):
    return {attr: target_dict.get(attr) for attr in attributes if target_dict.get(attr)}

def _update_dict(target_dict, field, value):
    target_dict[field] = value
    return target_dict

def update_container_definitions(container_definitions, name, new_image):
    update_matching = [_update_dict(definition, 'image', new_image) 
            for definition in container_definitions if definition['name'] == name]
    
    filter_not_matching = [_update_dict(definition, 'image', new_image) 
            for definition in container_definitions if definition['name'] != name]
    
    return update_matching + filter_not_matching


def get_task_definition(client, task_definition_name):
    attributes = [
        'family',
        'taskRoleArn',
        'executionRoleArn',
        'networkMode',
        'volumes',
        'placementConstraints',
        'requiresCompatibilities',
        'cpu',
        'memory',
        'tags',
        'pidMode',
        'ipcMode',
        'proxyConfiguration',
        'containerDefinitions'
    ]
    
    task_definition_response = client.describe_task_definition(
        taskDefinition=task_definition_name
    )

    return _get_attributes(task_definition_response['taskDefinition'], attributes) 

def get_service(client, cluster_name, service_name):
    attributes = [
        'desiredCount',
        'deploymentConfiguration',
        'networkConfiguration',
        'platformVersion',
        'healthCheckGracePeriodSeconds'
    ]
        
    service_response = client.describe_services(
        cluster = cluster_name,
        services = [service_name]
    )

    return _get_attributes(service_response['services'][0], attributes) 
    
    
def register_task_definition(client, task_definition):
    register_response = client.register_task_definition(**task_definition)
    
    family = register_response['taskDefinition']['family']
    revision = register_response['taskDefinition']['revision']

    return f"{family}:{revision}"

def update_task_definition(client, task_definition_name, container_name, new_image):
    task_definition = get_task_definition(client, task_definition_name)
    container_definitions = task_definition['containerDefinitions']
    
    updated_container_definitions = update_container_definitions(
        container_definitions, 
        container_name, 
        new_image
    )
    
    task_definition['containerDefinitions'] = updated_container_definitions

    new_task_definition = register_task_definition(client, task_definition)
    
    return new_task_definition

def update_service(client, cluster_name, service_name, new_task_definition):
    service = get_service(client, cluster_name, service_name)
    
    service['cluster'] = cluster_name
    service['service'] = service_name
    service['taskDefinition'] = new_task_definition
    
    return client.update_service(**service)

def deploy_new_image(task_definition_name, container_name, image_name, cluster_name, service_name):
    role_arn="arn:aws:iam::975596993436:role/storage-developer"
    session = create_session(role_arn)
    client = session.client('ecs')

    new_task_definition = update_task_definition(
        client, 
        task_definition_name, 
        container_name, 
        image_name
    )
    
    update_service_response = update_service(
        client, 
        cluster_name, 
        service_name, 
        new_task_definition
    )
    
    return {
        'created_task_definition': new_task_definition,
        'deployed_task_definition': update_service_response['service']['taskDefinition']
    }

# # Usage:
# deployment = {
#     'task_definition_name': 'existing_task_definition_name',
#     'container_name': 'app',
#     'image_name': 'hello_world:latest',
#     'cluster_name': 'staging_cluster',
#     'service_name': 'my_service'
# }
#
# update_task_definition(client, task_definition_name, container_name, image_name)