ecs_deploy.EcsAction.service_name() - Code Metrics - Inspection of "Change strings to unicode and add some more tests" - fabfuel/ecs-deploy - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — develop ( 0996c4...7f4e8d )

by Fabian

created 2016-03-15 07:40 UTC

ecs_deploy.EcsAction.service_name() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	1

Metric	Value
dl	0
loc	3
ccs	2
cts	2
cp	1
rs	10
cc	1
crap	1

from datetime import datetime

import boto3
from botocore.exceptions import ClientError, NoCredentialsError


class EcsClient(object):
    def __init__(self, access_key_id=None, secret_access_key=None, region=None, profile=None):
        session = boto3.session.Session(aws_access_key_id=access_key_id,
                                        aws_secret_access_key=secret_access_key,
                                        region_name=region,
                                        profile_name=profile)
        self.boto = session.client(u'ecs')

    def describe_services(self, cluster_name, service_name):
        return self.boto.describe_services(cluster=cluster_name, services=[service_name])

    def describe_task_definition(self, task_definition_arn):
        return self.boto.describe_task_definition(taskDefinition=task_definition_arn)

    def list_tasks(self, cluster_name, service_name):
        return self.boto.list_tasks(cluster=cluster_name, serviceName=service_name)

    def describe_tasks(self, cluster_name, task_arns):
        return self.boto.describe_tasks(cluster=cluster_name, tasks=task_arns)

    def register_task_definition(self, family, containers, volumes):
        return self.boto.register_task_definition(family=family, containerDefinitions=containers, volumes=volumes)

    def deregister_task_definition(self, task_definition_arn):
        return self.boto.deregister_task_definition(taskDefinition=task_definition_arn)

    def update_service(self, cluster, service, desired_count, task_definition):
        return self.boto.update_service(
            cluster=cluster,
            service=service,
            desiredCount=desired_count,
            taskDefinition=task_definition
        )


class EcsService(dict):
    def __init__(self, cluster, iterable=None, **kwargs):
        self._cluster = cluster
        super(EcsService, self).__init__(iterable, **kwargs)

    def set_desired_count(self, desired_count):
        self[u'desiredCount'] = desired_count

    def set_task_definition(self, task_definition):
        self[u'taskDefinition'] = task_definition.arn

    @property
    def cluster(self):
        return self._cluster

    @property
    def name(self):
        return self.get(u'serviceName')

    @property
    def task_definition(self):
        return self.get(u'taskDefinition')

    @property
    def desired_count(self):
        return self.get(u'desiredCount')

    @property
    def deployment_created_at(self):
        for deployment in self.get(u'deployments'):
            if deployment.get(u'status') == u'PRIMARY':
                return deployment.get(u'createdAt')
        return datetime.now()

    @property
    def deployment_updated_at(self):
        for deployment in self.get(u'deployments'):
            if deployment.get(u'status') == u'PRIMARY':
                return deployment.get(u'updatedAt')
        return datetime.now()


class EcsTaskDefinition(dict):
    def __init__(self, iterable=None, **kwargs):
        super(EcsTaskDefinition, self).__init__(iterable, **kwargs)
        self._diff = []

    @property
    def containers(self):
        return self.get(u'containerDefinitions')

    @property
    def container_names(self):
        for container in self.get(u'containerDefinitions'):
            yield container[u'name']

    @property
    def volumes(self):
        return self.get(u'volumes')

    @property
    def arn(self):
        return self.get(u'taskDefinitionArn')

    @property
    def family(self):
        return self.get(u'family')

    @property
    def revision(self):
        return self.get(u'revision')

    @property
    def diff(self):
        return self._diff

    def set_images(self, tag=None, **images):
        self.validate_image_options(**images)
        for container in self.containers:
            if container[u'name'] in images:
                new_image = images[container[u'name']]
                diff = EcsTaskDefinitionDiff(container[u'name'], u'image', new_image, container[u'image'])
                self._diff.append(diff)
                container[u'image'] = new_image
            elif tag:
                image_definition = container[u'image'].rsplit(u':', 1)
                new_image = u'%s:%s' % (image_definition[0], tag)
                diff = EcsTaskDefinitionDiff(container[u'name'], u'image', new_image, container[u'image'])
                self._diff.append(diff)
                container[u'image'] = new_image

    def set_commands(self, **commands):
        self.validate_image_options(**commands)
        for container in self.containers:
            if container[u'name'] in commands:
                new_command = commands[container[u'name']]
                diff = EcsTaskDefinitionDiff(container[u'name'], u'command', new_command, container.get(u'command'))
                self._diff.append(diff)
                container[u'command'] = [new_command]

    def validate_image_options(self, **container_options):
        for container_name in container_options:
            if container_name not in self.container_names:
                raise UnknownContainerError(u'Unknown container: %s' % container_name)


class EcsTaskDefinitionDiff(object):
    def __init__(self, container, field, value, old_value):
        self.container = container
        self.field = field
        self.value = value
        self.old_value = old_value

    def __repr__(self):
        return u"Changed %s of container '%s' to: %s (was: %s)" % \
               (self.field, self.container, self.value, self.old_value)


class EcsAction(object):
    def __init__(self, client, cluster, service):
        self._client = client
        self._cluster_name = cluster
        self._service_name = service

        try:
            self._service = self.get_service()
        except IndexError:
            raise ConnectionError(u'An error occurred when calling the DescribeServices operation: Service not found')
        except ClientError as e:
            raise ConnectionError(str(e))
        except NoCredentialsError:
            raise ConnectionError(u'Unable to locate credentials. Configure credentials by running "aws configure".')

    def get_service(self):
        services_definition = self._client.describe_services(self._cluster_name, self._service_name)
        return EcsService(self._cluster_name, services_definition[u'services'][0])

    def get_current_task_definition(self):
        task_definition = self._client.describe_task_definition(self._service.task_definition)
        task_definition = EcsTaskDefinition(task_definition[u'taskDefinition'])
        return task_definition

    def update_task_definition(self, task_definition):
        response = self._client.register_task_definition(task_definition.family, task_definition.containers,
                                                         task_definition.volumes)
        new_task_definition = EcsTaskDefinition(response[u'taskDefinition'])
        self._client.deregister_task_definition(task_definition.arn)
        return new_task_definition

    def update_service(self, service):
        response = self._client.update_service(service.cluster, service.name, service.desired_count,
                                               service.task_definition)
        return EcsService(self._cluster_name, response[u'service'])

    def is_deployed(self):
        running_tasks = self._client.list_tasks(self._cluster_name, self._service_name)
        if not running_tasks[u'taskArns']:
            return self._service.desired_count == 0
        return self._service.desired_count == self.get_running_tasks_count(running_tasks[u'taskArns'])

    def get_running_tasks_count(self, task_arns):
        running_count = 0
        tasks_details = self._client.describe_tasks(self._cluster_name, task_arns)
        for task in tasks_details[u'tasks']:
            if task[u'taskDefinitionArn'] == self._service.task_definition and task[u'lastStatus'] == u'RUNNING':
                running_count += 1
        return running_count

    @property
    def client(self):
        return self._client

    @property
    def service(self):
        return self._service

    @property
    def cluster_name(self):
        return self._cluster_name

    @property
    def service_name(self):
        return self._service_name


class DeployAction(EcsAction):
    def deploy(self, task_definition):
        self._service.set_task_definition(task_definition)
        return self.update_service(self._service)


class ScaleAction(EcsAction):
    def scale(self, desired_count):
        self._service.set_desired_count(desired_count)
        return self.update_service(self._service)


class EcsError(Exception):
    pass


class ConnectionError(EcsError):
    pass


class UnknownContainerError(EcsError):
    pass


1	1	from datetime import datetime
2
3	1	import boto3
4	1	from botocore.exceptions import ClientError, NoCredentialsError
5
6
7	1	class EcsClient(object):
8	1	def __init__(self, access_key_id=None, secret_access_key=None, region=None, profile=None):
9	1	session = boto3.session.Session(aws_access_key_id=access_key_id,
10		aws_secret_access_key=secret_access_key,
11		region_name=region,
12		profile_name=profile)
13	1	self.boto = session.client(u'ecs')
14
15	1	def describe_services(self, cluster_name, service_name):
16	1	return self.boto.describe_services(cluster=cluster_name, services=[service_name])
17
18	1	def describe_task_definition(self, task_definition_arn):
19	1	return self.boto.describe_task_definition(taskDefinition=task_definition_arn)
20
21	1	def list_tasks(self, cluster_name, service_name):
22	1	return self.boto.list_tasks(cluster=cluster_name, serviceName=service_name)
23
24	1	def describe_tasks(self, cluster_name, task_arns):
25	1	return self.boto.describe_tasks(cluster=cluster_name, tasks=task_arns)
26
27	1	def register_task_definition(self, family, containers, volumes):
28	1	return self.boto.register_task_definition(family=family, containerDefinitions=containers, volumes=volumes)
29
30	1	def deregister_task_definition(self, task_definition_arn):
31	1	return self.boto.deregister_task_definition(taskDefinition=task_definition_arn)
32
33	1	def update_service(self, cluster, service, desired_count, task_definition):
34	1	return self.boto.update_service(
35		cluster=cluster,
36		service=service,
37		desiredCount=desired_count,
38		taskDefinition=task_definition
39		)
40
41
42	1	class EcsService(dict):
43	1	def __init__(self, cluster, iterable=None, **kwargs):
44	1	self._cluster = cluster
45	1	super(EcsService, self).__init__(iterable, **kwargs)
46
47	1	def set_desired_count(self, desired_count):
48	1	self[u'desiredCount'] = desired_count
49
50	1	def set_task_definition(self, task_definition):
51	1	self[u'taskDefinition'] = task_definition.arn
52
53	1	@property
54		def cluster(self):
55	1	return self._cluster
56
57	1	@property
58		def name(self):
59	1	return self.get(u'serviceName')
60
61	1	@property
62		def task_definition(self):
63	1	return self.get(u'taskDefinition')
64
65	1	@property
66		def desired_count(self):
67	1	return self.get(u'desiredCount')
68
69	1	@property
70		def deployment_created_at(self):
71	1	for deployment in self.get(u'deployments'):
72	1	if deployment.get(u'status') == u'PRIMARY':
73	1	return deployment.get(u'createdAt')
74	1	return datetime.now()
75
76	1	@property
77		def deployment_updated_at(self):
78	1	for deployment in self.get(u'deployments'):
79	1	if deployment.get(u'status') == u'PRIMARY':
80	1	return deployment.get(u'updatedAt')
81	1	return datetime.now()
82
83
84	1	class EcsTaskDefinition(dict):
85	1	def __init__(self, iterable=None, **kwargs):
86	1	super(EcsTaskDefinition, self).__init__(iterable, **kwargs)
87	1	self._diff = []
88
89	1	@property
90		def containers(self):
91	1	return self.get(u'containerDefinitions')
92
93	1	@property
94		def container_names(self):
95	1	for container in self.get(u'containerDefinitions'):
96	1	yield container[u'name']
97
98	1	@property
99		def volumes(self):
100	1	return self.get(u'volumes')
101
102	1	@property
103		def arn(self):
104	1	return self.get(u'taskDefinitionArn')
105
106	1	@property
107		def family(self):
108	1	return self.get(u'family')
109
110	1	@property
111		def revision(self):
112	1	return self.get(u'revision')
113
114	1	@property
115		def diff(self):
116	1	return self._diff
117
118	1	def set_images(self, tag=None, **images):
119	1	self.validate_image_options(**images)
120	1	for container in self.containers:
121	1	if container[u'name'] in images:
122	1	new_image = images[container[u'name']]
123	1	diff = EcsTaskDefinitionDiff(container[u'name'], u'image', new_image, container[u'image'])
124	1	self._diff.append(diff)
125	1	container[u'image'] = new_image
126	1	elif tag:
127	1	image_definition = container[u'image'].rsplit(u':', 1)
128	1	new_image = u'%s:%s' % (image_definition[0], tag)
129	1	diff = EcsTaskDefinitionDiff(container[u'name'], u'image', new_image, container[u'image'])
130	1	self._diff.append(diff)
131	1	container[u'image'] = new_image
132
133	1	def set_commands(self, **commands):
134	1	self.validate_image_options(**commands)
135	1	for container in self.containers:
136	1	if container[u'name'] in commands:
137	1	new_command = commands[container[u'name']]
138	1	diff = EcsTaskDefinitionDiff(container[u'name'], u'command', new_command, container.get(u'command'))
139	1	self._diff.append(diff)
140	1	container[u'command'] = [new_command]
141
142	1	def validate_image_options(self, **container_options):
143	1	for container_name in container_options:
144	1	if container_name not in self.container_names:
145	1	raise UnknownContainerError(u'Unknown container: %s' % container_name)
146
147
148	1	class EcsTaskDefinitionDiff(object):
149	1	def __init__(self, container, field, value, old_value):
150	1	self.container = container
151	1	self.field = field
152	1	self.value = value
153	1	self.old_value = old_value
154
155	1	def __repr__(self):
156	1	return u"Changed %s of container '%s' to: %s (was: %s)" % \
157		(self.field, self.container, self.value, self.old_value)
158
159
160	1	class EcsAction(object):
161	1	def __init__(self, client, cluster, service):
162	1	self._client = client
163	1	self._cluster_name = cluster
164	1	self._service_name = service
165
166	1	try:
167	1	self._service = self.get_service()
168	1	except IndexError:
169	1	raise ConnectionError(u'An error occurred when calling the DescribeServices operation: Service not found')
170	1	except ClientError as e:
171	1	raise ConnectionError(str(e))
172	1	except NoCredentialsError:
173	1	raise ConnectionError(u'Unable to locate credentials. Configure credentials by running "aws configure".')
174
175	1	def get_service(self):
176	1	services_definition = self._client.describe_services(self._cluster_name, self._service_name)
177	1	return EcsService(self._cluster_name, services_definition[u'services'][0])
178
179	1	def get_current_task_definition(self):
180	1	task_definition = self._client.describe_task_definition(self._service.task_definition)
181	1	task_definition = EcsTaskDefinition(task_definition[u'taskDefinition'])
182	1	return task_definition
183
184	1	def update_task_definition(self, task_definition):
185		response = self._client.register_task_definition(task_definition.family, task_definition.containers,
186		task_definition.volumes)
187		new_task_definition = EcsTaskDefinition(response[u'taskDefinition'])
188		self._client.deregister_task_definition(task_definition.arn)
189		return new_task_definition
190
191	1	def update_service(self, service):
192		response = self._client.update_service(service.cluster, service.name, service.desired_count,
193		service.task_definition)
194		return EcsService(self._cluster_name, response[u'service'])
195
196	1	def is_deployed(self):
197		running_tasks = self._client.list_tasks(self._cluster_name, self._service_name)
198		if not running_tasks[u'taskArns']:
199		return self._service.desired_count == 0
200		return self._service.desired_count == self.get_running_tasks_count(running_tasks[u'taskArns'])
201
202	1	def get_running_tasks_count(self, task_arns):
203		running_count = 0
204		tasks_details = self._client.describe_tasks(self._cluster_name, task_arns)
205		for task in tasks_details[u'tasks']:
206		if task[u'taskDefinitionArn'] == self._service.task_definition and task[u'lastStatus'] == u'RUNNING':
207		running_count += 1
208		return running_count
209
210	1	@property
211		def client(self):
212	1	return self._client
213
214	1	@property
215		def service(self):
216	1	return self._service
217
218	1	@property
219		def cluster_name(self):
220	1	return self._cluster_name
221
222	1	@property
223		def service_name(self):
224	1	return self._service_name
225
226
227	1	class DeployAction(EcsAction):
228	1	def deploy(self, task_definition):
229		self._service.set_task_definition(task_definition)
230		return self.update_service(self._service)
231
232
233	1	class ScaleAction(EcsAction):
234	1	def scale(self, desired_count):
235		self._service.set_desired_count(desired_count)
236		return self.update_service(self._service)
237
238
239	1	class EcsError(Exception):
240	1	pass
241
242
243	1	class ConnectionError(EcsError):
244	1	pass
245
246
247	1	class UnknownContainerError(EcsError):
248		pass
249

fabfuel / ecs-deploy

Push — develop ( 0996c4...7f4e8d )

ecs_deploy.EcsAction.service_name() A

Complexity

Size

Duplication

Code Coverage

Duplication Side-by-Side

Filter issues like