API reference

Deployment utilities

Shared library to help surface ECS deployment information.

class wellcome_aws_utils.deployment_utils.Deployment(deployment_key, deployment_status, color, created_at, task_definition)

color

Alias for field number 2

created_at

Alias for field number 3

deployment_key

Alias for field number 0

deployment_status

Alias for field number 1

task_definition

Alias for field number 4

class wellcome_aws_utils.deployment_utils.DeploymentKey(id, service_arn)

id

Alias for field number 0

service_arn

Alias for field number 1

DynamoDB events

class wellcome_aws_utils.dynamo_event.DynamoEventType

An enumeration.

DynamoDB utilities

wellcome_aws_utils.dynamo_utils.change_dynamo_capacity(client, table_name, desired_capacity)

Given the name of a DynamoDB table and a desired capacity, update the read/write capacity of the table and every secondary index.

ECS

exception wellcome_aws_utils.ecs_utils.EcsThrottleException

wellcome_aws_utils.ecs_utils.clone_task_definition(client, task_definition)

Given a task definition ARN, clone the associated task.

Returns the new task definition ARN.

wellcome_aws_utils.ecs_utils.describe_cluster(ecs_client, cluster_arn)

Given a cluster ARN attempts to find a matching cluster description.

Returns a cluster description.

wellcome_aws_utils.ecs_utils.describe_service(ecs_client, cluster_arn, service_arn)

Given a cluster ARN and service ARN, attempts to find a matching service description.

Returns a service description.

wellcome_aws_utils.ecs_utils.get_cluster_arns(ecs_client)

Extract the list of cluster ARNs in this account.

Returns a list of cluster ARNs.

wellcome_aws_utils.ecs_utils.get_latest_task_definition(client, cluster, service)

Given the name of a cluster and a service, return the ARN for its latest task definition.

wellcome_aws_utils.ecs_utils.get_service_arns(ecs_client, cluster_arn)

Given a cluster ARN, extracts the associated service ARNs.

Returns a list of service ARNS.

wellcome_aws_utils.ecs_utils.identify_cluster_by_app_name(client, app_name)

Given the name of one of our applications (e.g. api, calm_adapter), return the ARN of the cluster the task runs on.

wellcome_aws_utils.ecs_utils.run_task(ecs_client, cluster_name, task_definition, started_by, container_name='app', command=[])

Run a given command against a named container in a task definition on a particular cluster.

Returns the response from calling run_task

S3

wellcome_aws_utils.s3_utils.is_object(bucket, key)

Checks if an object exists in S3. Returns True/False.

Parameters:

• bucket – Bucket of the object to check.
• key – Key of the object to check.

wellcome_aws_utils.s3_utils.copy_object(src_bucket, src_key, dst_bucket, dst_key, lazy=False)

Copy an object from one S3 bucket to another.

Parameters:

• src_bucket – Bucket of the source object.
• src_key – Key of the source object.
• dst_bucket – Bucket of the destination object.
• dst_key – Key of the destination object.
• lazy – Do a lazy copy. This means that the object will only be copied if the destination object does not exist, or exists but has a different ETag from the source object.

wellcome_aws_utils.s3_utils.parse_s3_record(event)

Extracts a simple subset of an S3 update event.

wellcome_aws_utils.s3_utils.write_objects_to_s3(bucket, key, objects)

Given an iterable of objects that can be serialised as JSON, serialise them as JSON, and write them to a file in S3, one per line.

Parameters:

• bucket – S3 bucket to upload the new file to.
• key – S3 key to upload the new file to.
• objects – An iterable of objects that can be serialised as JSON.

SNS

class wellcome_aws_utils.sns_utils.EnhancedJSONEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)

default(obj)

Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this:

def default(self, o):
    try:
        iterable = iter(o)
    except TypeError:
        pass
    else:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return JSONEncoder.default(self, o)

class wellcome_aws_utils.sns_utils.SNSEvent(subject, message)

message

Alias for field number 1

subject

Alias for field number 0

wellcome_aws_utils.sns_utils.extract_json_message(event)

Extracts a JSON message from an SNS event sent to a lambda

Deprecated in favour of extract_sns_messages_from_lambda_event

wellcome_aws_utils.sns_utils.extract_sns_messages_from_lambda_event(event)

Extracts a JSON message from an SNS event sent to an AWS Lambda.

Parameters:event – An event sent to a Lambda from SNS. Returns:A generator of SNSEvent instances.

wellcome_aws_utils.sns_utils.publish_sns_message(sns_client, topic_arn, message, subject='default-subject')

Given a topic ARN and a series of key-value pairs, publish the key-value data to the SNS topic.

SQS

wellcome_aws_utils.sqs_utils.get_messages(queue_url, delete=False, batch_size=10)

Gets messages from an SQS queue. If delete is True, the messages are also deleted after they’ve been read.