Created
August 19, 2018 12:19
-
-
Save fbidu/942d87da28f054f96dfcc4b40cb1638b to your computer and use it in GitHub Desktop.
A rough draft for a more pythonic AWS client
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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