Skip to content

Instantly share code, notes, and snippets.

@fbidu
Created August 19, 2018 12:19
Show Gist options
  • Save fbidu/942d87da28f054f96dfcc4b40cb1638b to your computer and use it in GitHub Desktop.
Save fbidu/942d87da28f054f96dfcc4b40cb1638b to your computer and use it in GitHub Desktop.
A rough draft for a more pythonic AWS client
import boto3
class Cluster:
"""
Class to handle an ECS Cluster
"""
def __init__(self, name, boto3_session):
assert type(boto3_session) == boto3.session.Session, "Please, provide an boto3 Session as argument!"
self.name = name
self.boto3_session = boto3_session
self.ecs = boto3_session.client('ecs')
self.__services = []
@property
def services(self):
"""
Lists all the services in a cluster
"""
if not self.__services:
data = self.ecs.list_services(cluster=self.name)
for item in data['serviceArns']:
self.__services.append(Service(item, self, self.boto3_session))
return self.__services
class Service:
"""
Class to handle ECS Services
"""
__description = None
__task_definition = None
__tasks = []
def __init__(self, name, cluster, boto3_session):
assert type(boto3_session) == boto3.session.Session, "Please, provide an boto3 Session as argument!"
self.name = name
self.cluster = cluster
self.ecs = boto3_session.client('ecs')
def update_description(self):
"""
Calls AWS's describe_services (http://bit.ly/2OHJqm8) to get more data
about the current service
"""
data = self.ecs.describe_services(cluster=self.cluster.name, services=[self.name])
try:
# Do we have data?
assert data
# Is it in the format we expect?
assert type(data) == dict
# Do we have failures?
assert len(data['failures']) == 0
# Does the metadata points to a successful request?
assert int(data['ResponseMetadata']['HTTPStatusCode']) < 400
# Is there actual data in it and do we have exactly one service?
assert len(data['services']) == 1
except AssertionError as e:
print("Found an assertion error {}".format(e))
print("The current state of failures is: ")
print("\n".join(data['failures']))
else:
self.__description = data['services'][0]
@property
def task_definition(self):
if not self.__task_definition:
if not self.__description:
self.update_description()
self.__task_definition = self.__description['taskDefinition']
return self.__task_definition
def main():
session = boto3.Session()
cluster = Cluster("MyCluster", session)
for service in cluster.services:
print(service.task_definition)
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment