EcsTaskDefinition.set_images() - Code Metrics - Inspection of "Merge branch 'release/1.2.0'" - fabfuel/ecs-deploy - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 553be1...b5a472 )

by Fabian

created 2017-06-28 05:09 UTC

EcsTaskDefinition.set_images() B

↳ Parent: EcsTaskDefinition

Complexity

Conditions

Size

Total Lines

Duplication

Lines	11
Ratio	45.83 %

Code Coverage

Tests	14
CRAP Score	4

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
c	1
b	0
f	0
dl	11
loc	24
ccs	14
cts	14
cp	1
rs	8.6845
cc	4
crap	4

from datetime import datetime
from json import dumps

from boto3.session import Session
from botocore.exceptions import ClientError, NoCredentialsError
from dateutil.tz.tz import tzlocal


class EcsClient(object):
    def __init__(self, access_key_id=None, secret_access_key=None,
                 region=None, profile=None):
        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):
        try:
            return self.boto.describe_task_definition(
                taskDefinition=task_definition_arn
            )
        except ClientError:
            raise UnknownTaskDefinitionError(
                u'Unknown task definition arn: %s' % 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, role_arn):
        return self.boto.register_task_definition(
            family=family,
            containerDefinitions=containers,
            volumes=volumes,
            taskRoleArn=role_arn or u''
        )

    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
        )

    def run_task(self, cluster, task_definition, count, started_by, overrides):
        return self.boto.run_task(
            cluster=cluster,
            taskDefinition=task_definition,
            count=count,
            startedBy=started_by,
            overrides=overrides
        )


class EcsService(dict):
    def __init__(self, cluster, service_definition=None, **kwargs):
        self._cluster = cluster
        super(EcsService, self).__init__(service_definition, **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()

    @property
    def errors(self):
        return self.get_warnings(
            since=self.deployment_updated_at
        )

    @property
    def older_errors(self):
        return self.get_warnings(
            since=self.deployment_created_at,
            until=self.deployment_updated_at
        )

    def get_warnings(self, since=None, until=None):
        since = since or self.deployment_created_at
        until = until or datetime.now(tz=tzlocal())
        errors = {}
        for event in self.get(u'events'):
            if u'unable' not in event[u'message']:
                continue
            if since < event[u'createdAt'] < until:
                errors[event[u'createdAt']] = event[u'message']
        return errors


class EcsTaskDefinition(dict):
    def __init__(self, task_definition=None, **kwargs):
        super(EcsTaskDefinition, self).__init__(task_definition, **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 role_arn(self):
        return self.get(u'taskRoleArn')

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

    @property
    def family_revision(self):
        return '%s:%d' % (self.get(u'family'), self.get(u'revision'))

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

    def get_overrides(self):
        override = dict()
        overrides = []
        for diff in self.diff:
            if override.get('name') != diff.container:
                override = dict(name=diff.container)
                overrides.append(override)
            if diff.field == 'command':
                override['command'] = self.get_overrides_command(diff.value)
            elif diff.field == 'environment':
                override['environment'] = self.get_overrides_env(diff.value)
        return overrides

    @staticmethod
    def get_overrides_command(command):
        return command.split(' ')

    @staticmethod
    def get_overrides_env(env):
        return [{"name": e, "value": env[e]} for e in env]

    def set_images(self, tag=None, **images):
        self.validate_container_options(**images)
        for container in self.containers:
            if container[u'name'] in images:
                new_image = images[container[u'name']]
                diff = EcsTaskDefinitionDiff(
                    container=container[u'name'],
                    field=u'image',
                    value=new_image,
                    old_value=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.strip())
                diff = EcsTaskDefinitionDiff(
                    container=container[u'name'],
                    field=u'image',
                    value=new_image,
                    old_value=container[u'image']
                )
                self._diff.append(diff)
                container[u'image'] = new_image

    def set_commands(self, **commands):
        self.validate_container_options(**commands)
        for container in self.containers:
            if container[u'name'] in commands:

                new_command = commands[container[u'name']]
                diff = EcsTaskDefinitionDiff(
                    container=container[u'name'],
                    field=u'command',
                    value=new_command,
                    old_value=container.get(u'command')
                )
                self._diff.append(diff)
                container[u'command'] = [new_command]

    def set_environment(self, environment_list):
        environment = {}

        for env in environment_list:
            environment.setdefault(env[0], {})
            environment[env[0]][env[1]] = env[2]

        self.validate_container_options(**environment)
        for container in self.containers:
            if container[u'name'] in environment:
                self.apply_container_environment(
                    container=container,
                    new_environment=environment[container[u'name']]
                )

    def apply_container_environment(self, container, new_environment):
        environment = container.get('environment', {})
        old_environment = {env['name']: env['value'] for env in environment}
        merged = old_environment.copy()
        merged.update(new_environment)

        diff = EcsTaskDefinitionDiff(
            container=container[u'name'],
            field=u'environment',
            value=merged,
            old_value=old_environment
        )
        self._diff.append(diff)

        container[u'environment'] = [
            {"name": e, "value": merged[e]} for e in merged
        ]

    def validate_container_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
                )

    def set_role_arn(self, role_arn):
        if role_arn:
            diff = EcsTaskDefinitionDiff(
                container=None,
                field=u'role_arn',
                value=role_arn,
                old_value=self[u'taskRoleArn']
            )
            self[u'taskRoleArn'] = role_arn
            self._diff.append(diff)


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):
        if self.container:
            return u"Changed %s of container '%s' to: %s (was: %s)" % (
                self.field,
                self.container,
                dumps(self.value),
                dumps(self.old_value)
            )
        else:
            return u"Changed %s to: %s (was: %s)" % (
                self.field,
                dumps(self.value),
                dumps(self.old_value)
            )


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

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

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

    def get_current_task_definition(self, service):
        task_definition_payload = self._client.describe_task_definition(
            task_definition_arn=service.task_definition
        )
        task_definition = EcsTaskDefinition(
            task_definition=task_definition_payload[u'taskDefinition']
        )
        return task_definition

    def get_task_definition(self, task_definition):
        task_definition_payload = self._client.describe_task_definition(
            task_definition_arn=task_definition
        )
        task_definition = EcsTaskDefinition(
            task_definition=task_definition_payload[u'taskDefinition']
        )
        return task_definition

    def update_task_definition(self, task_definition):
        response = self._client.register_task_definition(
            family=task_definition.family,
            containers=task_definition.containers,
            volumes=task_definition.volumes,
            role_arn=task_definition.role_arn
        )
        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(
            cluster=service.cluster,
            service=service.name,
            desired_count=service.desired_count,
            task_definition=service.task_definition
        )
        return EcsService(self._cluster_name, response[u'service'])

    def is_deployed(self, service):
        if len(service[u'deployments']) != 1:
            return False
        running_tasks = self._client.list_tasks(
            cluster_name=service.cluster,
            service_name=service.name
        )
        if not running_tasks[u'taskArns']:
            return service.desired_count == 0
        running_count = self.get_running_tasks_count(
            service=service,
            task_arns=running_tasks[u'taskArns']
        )
        return service.desired_count == running_count

    def get_running_tasks_count(self, service, task_arns):
        running_count = 0
        tasks_details = self._client.describe_tasks(
            cluster_name=self._cluster_name,
            task_arns=task_arns
        )
        for task in tasks_details[u'tasks']:
            arn = task[u'taskDefinitionArn']
            status = task[u'lastStatus']
            if arn == service.task_definition and status == 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 RunAction(EcsAction):
    def __init__(self, client, cluster_name):
        super(RunAction, self).__init__(client, cluster_name, None)
        self._client = client
        self._cluster_name = cluster_name
        self.started_tasks = []

    def run(self, task_definition, count, started_by):
        result = self._client.run_task(
            cluster=self._cluster_name,
            task_definition=task_definition.family_revision,
            count=count,
            started_by=started_by,
            overrides=dict(containerOverrides=task_definition.get_overrides())
        )
        self.started_tasks = result['tasks']
        return True


class EcsError(Exception):
    pass


class EcsConnectionError(EcsError):
    pass


class UnknownContainerError(EcsError):
    pass


class TaskPlacementError(EcsError):
    pass


class UnknownTaskDefinitionError(EcsError):
    pass


1	1		from datetime import datetime
2	1		from json import dumps
3
4	1		from boto3.session import Session
5	1		from botocore.exceptions import ClientError, NoCredentialsError
6	1		from dateutil.tz.tz import tzlocal
7
8
9	1		class EcsClient(object):
10	1		def __init__(self, access_key_id=None, secret_access_key=None,
11			region=None, profile=None):
12	1		session = Session(aws_access_key_id=access_key_id,
13			aws_secret_access_key=secret_access_key,
14			region_name=region,
15			profile_name=profile)
16	1		self.boto = session.client(u'ecs')
17
18	1		def describe_services(self, cluster_name, service_name):
19	1		return self.boto.describe_services(
20			cluster=cluster_name,
21			services=[service_name]
22			)
23
24	1		def describe_task_definition(self, task_definition_arn):
25	1		try:
26	1		return self.boto.describe_task_definition(
27			taskDefinition=task_definition_arn
28			)
29	1		except ClientError:
30	1		raise UnknownTaskDefinitionError(
31			u'Unknown task definition arn: %s' % task_definition_arn
32			)
33
34	1		def list_tasks(self, cluster_name, service_name):
35	1		return self.boto.list_tasks(
36			cluster=cluster_name,
37			serviceName=service_name
38			)
39
40	1		def describe_tasks(self, cluster_name, task_arns):
41	1		return self.boto.describe_tasks(cluster=cluster_name, tasks=task_arns)
42
43	1		def register_task_definition(self, family, containers, volumes, role_arn):
44	1		return self.boto.register_task_definition(
45			family=family,
46			containerDefinitions=containers,
47			volumes=volumes,
48			taskRoleArn=role_arn or u''
49			)
50
51	1		def deregister_task_definition(self, task_definition_arn):
52	1		return self.boto.deregister_task_definition(
53			taskDefinition=task_definition_arn
54			)
55
56	1		def update_service(self, cluster, service, desired_count, task_definition):
57	1		return self.boto.update_service(
58			cluster=cluster,
59			service=service,
60			desiredCount=desired_count,
61			taskDefinition=task_definition
62			)
63
64	1		def run_task(self, cluster, task_definition, count, started_by, overrides):
65	1		return self.boto.run_task(
66			cluster=cluster,
67			taskDefinition=task_definition,
68			count=count,
69			startedBy=started_by,
70			overrides=overrides
71			)
72
73
74	1		class EcsService(dict):
75	1		def __init__(self, cluster, service_definition=None, **kwargs):
76	1		self._cluster = cluster
77	1		super(EcsService, self).__init__(service_definition, **kwargs)
78
79	1		def set_desired_count(self, desired_count):
80	1		self[u'desiredCount'] = desired_count
81
82	1		def set_task_definition(self, task_definition):
83	1		self[u'taskDefinition'] = task_definition.arn
84
85	1		@property
86			def cluster(self):
87	1		return self._cluster
88
89	1		@property
90			def name(self):
91	1		return self.get(u'serviceName')
92
93	1		@property
94			def task_definition(self):
95	1		return self.get(u'taskDefinition')
96
97	1		@property
98			def desired_count(self):
99	1		return self.get(u'desiredCount')
100
101	1		@property
102			def deployment_created_at(self):
103	1		for deployment in self.get(u'deployments'):
104	1		if deployment.get(u'status') == u'PRIMARY':
105	1		return deployment.get(u'createdAt')
106	1		return datetime.now()
107
108	1		@property
109			def deployment_updated_at(self):
110	1		for deployment in self.get(u'deployments'):
111	1		if deployment.get(u'status') == u'PRIMARY':
112	1		return deployment.get(u'updatedAt')
113	1		return datetime.now()
114
115	1		@property
116			def errors(self):
117	1		return self.get_warnings(
118			since=self.deployment_updated_at
119			)
120
121	1		@property
122			def older_errors(self):
123	1		return self.get_warnings(
124			since=self.deployment_created_at,
125			until=self.deployment_updated_at
126			)
127
128	1		def get_warnings(self, since=None, until=None):
129	1		since = since or self.deployment_created_at
130	1		until = until or datetime.now(tz=tzlocal())
131	1		errors = {}
132	1		for event in self.get(u'events'):
133	1		if u'unable' not in event[u'message']:
134			continue
135	1		if since < event[u'createdAt'] < until:
136	1		errors[event[u'createdAt']] = event[u'message']
137	1		return errors
138
139
140	1		class EcsTaskDefinition(dict):
141	1		def __init__(self, task_definition=None, **kwargs):
142	1		super(EcsTaskDefinition, self).__init__(task_definition, **kwargs)
143	1		self._diff = []
144
145	1		@property
146			def containers(self):
147	1		return self.get(u'containerDefinitions')
148
149	1		@property
150			def container_names(self):
151	1		for container in self.get(u'containerDefinitions'):
152	1		yield container[u'name']
153
154	1		@property
155			def volumes(self):
156	1		return self.get(u'volumes')
157
158	1		@property
159			def arn(self):
160	1		return self.get(u'taskDefinitionArn')
161
162	1		@property
163			def family(self):
164	1		return self.get(u'family')
165
166	1		@property
167			def role_arn(self):
168	1		return self.get(u'taskRoleArn')
169
170	1		@property
171			def revision(self):
172	1		return self.get(u'revision')
173
174	1		@property
175			def family_revision(self):
176	1		return '%s:%d' % (self.get(u'family'), self.get(u'revision'))
177
178	1		@property
179			def diff(self):
180	1		return self._diff
181
182	1		def get_overrides(self):
183	1		override = dict()
184	1		overrides = []
185	1		for diff in self.diff:
186	1		if override.get('name') != diff.container:
187	1		override = dict(name=diff.container)
188	1		overrides.append(override)
189	1		if diff.field == 'command':
190	1		override['command'] = self.get_overrides_command(diff.value)
191	1		elif diff.field == 'environment':
192	1		override['environment'] = self.get_overrides_env(diff.value)
193	1		return overrides
194
195	1		@staticmethod
196			def get_overrides_command(command):
197	1		return command.split(' ')
198
199	1		@staticmethod
200			def get_overrides_env(env):
201	1		return [{"name": e, "value": env[e]} for e in env]
202
203	1		def set_images(self, tag=None, **images):
204	1		self.validate_container_options(**images)
205	1		for container in self.containers:
206	1		if container[u'name'] in images:
207	1		new_image = images[container[u'name']]
208	1		diff = EcsTaskDefinitionDiff(
209			container=container[u'name'],
210			field=u'image',
211			value=new_image,
212			old_value=container[u'image']
213			)
214	1		self._diff.append(diff)
215	1		container[u'image'] = new_image
216	1	View Code Duplication	elif tag:
			0 ignored issues – show Duplication introduced 2017-06-26 06:03 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
217	1		image_definition = container[u'image'].rsplit(u':', 1)
218	1		new_image = u'%s:%s' % (image_definition[0], tag.strip())
219	1		diff = EcsTaskDefinitionDiff(
220			container=container[u'name'],
221			field=u'image',
222			value=new_image,
223			old_value=container[u'image']
224			)
225	1		self._diff.append(diff)
226	1		container[u'image'] = new_image
227
228	1		def set_commands(self, **commands):
229	1		self.validate_container_options(**commands)
230	1		for container in self.containers:
231	1	View Code Duplication	if container[u'name'] in commands:
			0 ignored issues – show Duplication introduced 2017-06-26 06:03 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
232	1		new_command = commands[container[u'name']]
233	1		diff = EcsTaskDefinitionDiff(
234			container=container[u'name'],
235			field=u'command',
236			value=new_command,
237			old_value=container.get(u'command')
238			)
239	1		self._diff.append(diff)
240	1		container[u'command'] = [new_command]
241
242	1		def set_environment(self, environment_list):
243	1		environment = {}
244
245	1		for env in environment_list:
246	1		environment.setdefault(env[0], {})
247	1		environment[env[0]][env[1]] = env[2]
248
249	1		self.validate_container_options(**environment)
250	1		for container in self.containers:
251	1		if container[u'name'] in environment:
252	1		self.apply_container_environment(
253			container=container,
254			new_environment=environment[container[u'name']]
255			)
256
257	1		def apply_container_environment(self, container, new_environment):
258	1		environment = container.get('environment', {})
259	1		old_environment = {env['name']: env['value'] for env in environment}
260	1		merged = old_environment.copy()
261	1		merged.update(new_environment)
262
263	1		diff = EcsTaskDefinitionDiff(
264			container=container[u'name'],
265			field=u'environment',
266			value=merged,
267			old_value=old_environment
268			)
269	1		self._diff.append(diff)
270
271	1		container[u'environment'] = [
272			{"name": e, "value": merged[e]} for e in merged
273			]
274
275	1		def validate_container_options(self, **container_options):
276	1		for container_name in container_options:
277	1		if container_name not in self.container_names:
278	1		raise UnknownContainerError(
279			u'Unknown container: %s' % container_name
280			)
281
282	1		def set_role_arn(self, role_arn):
283	1		if role_arn:
284	1		diff = EcsTaskDefinitionDiff(
285			container=None,
286			field=u'role_arn',
287			value=role_arn,
288			old_value=self[u'taskRoleArn']
289			)
290	1		self[u'taskRoleArn'] = role_arn
291	1		self._diff.append(diff)
292
293
294	1		class EcsTaskDefinitionDiff(object):
295	1		def __init__(self, container, field, value, old_value):
296	1		self.container = container
297	1		self.field = field
298	1		self.value = value
299	1		self.old_value = old_value
300
301	1		def __repr__(self):
302	1		if self.container:
303	1		return u"Changed %s of container '%s' to: %s (was: %s)" % (
304			self.field,
305			self.container,
306			dumps(self.value),
307			dumps(self.old_value)
308			)
309			else:
310	1		return u"Changed %s to: %s (was: %s)" % (
311			self.field,
312			dumps(self.value),
313			dumps(self.old_value)
314			)
315
316
317	1		class EcsAction(object):
318	1		def __init__(self, client, cluster_name, service_name):
319	1		self._client = client
320	1		self._cluster_name = cluster_name
321	1		self._service_name = service_name
322
323	1		try:
324	1		if service_name:
325	1		self._service = self.get_service()
326	1		except IndexError:
327	1		raise EcsConnectionError(
328			u'An error occurred when calling the DescribeServices '
329			u'operation: Service not found.'
330			)
331	1		except ClientError as e:
332	1		raise EcsConnectionError(str(e))
333	1		except NoCredentialsError:
334	1		raise EcsConnectionError(
335			u'Unable to locate credentials. Configure credentials '
336			u'by running "aws configure".'
337			)
338
339	1		def get_service(self):
340	1		services_definition = self._client.describe_services(
341			cluster_name=self._cluster_name,
342			service_name=self._service_name
343			)
344	1		return EcsService(
345			cluster=self._cluster_name,
346			service_definition=services_definition[u'services'][0]
347			)
348
349	1		def get_current_task_definition(self, service):
350	1		task_definition_payload = self._client.describe_task_definition(
351			task_definition_arn=service.task_definition
352			)
353	1		task_definition = EcsTaskDefinition(
354			task_definition=task_definition_payload[u'taskDefinition']
355			)
356	1		return task_definition
357
358	1		def get_task_definition(self, task_definition):
359	1		task_definition_payload = self._client.describe_task_definition(
360			task_definition_arn=task_definition
361			)
362	1		task_definition = EcsTaskDefinition(
363			task_definition=task_definition_payload[u'taskDefinition']
364			)
365	1		return task_definition
366
367	1		def update_task_definition(self, task_definition):
368	1		response = self._client.register_task_definition(
369			family=task_definition.family,
370			containers=task_definition.containers,
371			volumes=task_definition.volumes,
372			role_arn=task_definition.role_arn
373			)
374	1		new_task_definition = EcsTaskDefinition(response[u'taskDefinition'])
375	1		self._client.deregister_task_definition(task_definition.arn)
376	1		return new_task_definition
377
378	1		def update_service(self, service):
379	1		response = self._client.update_service(
380			cluster=service.cluster,
381			service=service.name,
382			desired_count=service.desired_count,
383			task_definition=service.task_definition
384			)
385	1		return EcsService(self._cluster_name, response[u'service'])
386
387	1		def is_deployed(self, service):
388	1		if len(service[u'deployments']) != 1:
389	1		return False
390	1		running_tasks = self._client.list_tasks(
391			cluster_name=service.cluster,
392			service_name=service.name
393			)
394	1		if not running_tasks[u'taskArns']:
395	1		return service.desired_count == 0
396	1		running_count = self.get_running_tasks_count(
397			service=service,
398			task_arns=running_tasks[u'taskArns']
399			)
400	1		return service.desired_count == running_count
401
402	1		def get_running_tasks_count(self, service, task_arns):
403	1		running_count = 0
404	1		tasks_details = self._client.describe_tasks(
405			cluster_name=self._cluster_name,
406			task_arns=task_arns
407			)
408	1		for task in tasks_details[u'tasks']:
409	1		arn = task[u'taskDefinitionArn']
410	1		status = task[u'lastStatus']
411	1		if arn == service.task_definition and status == u'RUNNING':
412	1		running_count += 1
413	1		return running_count
414
415	1		@property
416			def client(self):
417	1		return self._client
418
419	1		@property
420			def service(self):
421	1		return self._service
422
423	1		@property
424			def cluster_name(self):
425	1		return self._cluster_name
426
427	1		@property
428			def service_name(self):
429	1		return self._service_name
430
431
432	1		class DeployAction(EcsAction):
433	1		def deploy(self, task_definition):
434	1		self._service.set_task_definition(task_definition)
435	1		return self.update_service(self._service)
436
437
438	1		class ScaleAction(EcsAction):
439	1		def scale(self, desired_count):
440	1		self._service.set_desired_count(desired_count)
441	1		return self.update_service(self._service)
442
443
444	1		class RunAction(EcsAction):
445	1		def __init__(self, client, cluster_name):
446	1		super(RunAction, self).__init__(client, cluster_name, None)
447	1		self._client = client
448	1		self._cluster_name = cluster_name
449	1		self.started_tasks = []
450
451	1		def run(self, task_definition, count, started_by):
452	1		result = self._client.run_task(
453			cluster=self._cluster_name,
454			task_definition=task_definition.family_revision,
455			count=count,
456			started_by=started_by,
457			overrides=dict(containerOverrides=task_definition.get_overrides())
458			)
459	1		self.started_tasks = result['tasks']
460	1		return True
461
462
463	1		class EcsError(Exception):
464	1		pass
465
466
467	1		class EcsConnectionError(EcsError):
468	1		pass
469
470
471	1		class UnknownContainerError(EcsError):
472	1		pass
473
474
475	1		class TaskPlacementError(EcsError):
476	1		pass
477
478
479	1		class UnknownTaskDefinitionError(EcsError):
480			pass
481

fabfuel / ecs-deploy

Push — master ( 553be1...b5a472 )

EcsTaskDefinition.set_images() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like