Update an ECS task definition image only

update_task_definition.py

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 register_task_definition(client, task_definition):
    register_response = client.register_task_definition(**task_definition)
    
    family = response['taskDefinition']['family']
    revision = 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

# # Usage:
# role_arn="arn:aws:iam::123456789:role/my_role"
# session = create_session(role_arn)
# client = session.client('ecs')
#
# task_definition_name="some_definition"
# container_name = 'app'
# image_name = 'plof/meefbot'
#
# update_task_definition(client, task_definition_name, container_name, image_name)