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)

Alias for field number 2


Alias for field number 3


Alias for field number 0


Alias for field number 1


Alias for field number 4

class wellcome_aws_utils.deployment_utils.DeploymentKey(id, service_arn)

Alias for field number 0


Alias for field number 1

DynamoDB events

class wellcome_aws_utils.dynamo_event.DynamoEventType[source]

An enumeration.

DynamoDB utilities

wellcome_aws_utils.dynamo_utils.change_dynamo_capacity(client, table_name, desired_capacity)[source]

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


exception wellcome_aws_utils.ecs_utils.EcsThrottleException[source]
wellcome_aws_utils.ecs_utils.clone_task_definition(client, task_definition)[source]

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)[source]

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)[source]

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

Returns a service description.


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)[source]

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)[source]

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)[source]

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=[])[source]

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

Returns the response from calling run_task


wellcome_aws_utils.s3_utils.is_object(bucket, key)[source]

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

  • 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)[source]

Copy an object from one S3 bucket to another.

  • 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.

Extracts a simple subset of an S3 update event.

wellcome_aws_utils.s3_utils.write_objects_to_s3(bucket, key, objects)[source]

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.

  • 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.


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)[source]

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):
        iterable = iter(o)
    except TypeError:
        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)

Alias for field number 1


Alias for field number 0


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

Deprecated in favour of extract_sns_messages_from_lambda_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')[source]

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


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

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