Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Use list comprehensions instead of filter() for easier moto/swf python 3.x compatibility

Browse Source
This commit is contained in:
Jean-Baptiste Barth 2015-11-03 10:45:11 +01:00
parent 3a5f679783
commit b386495520
2 changed files with 7 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
moto/swf/models/__init__.py
View File

@ -193,7 +193,7 @@ class SWFBackend(BaseBackend):
candidates = [] candidates = []
for _task_list, tasks in domain.decision_task_lists.items(): for _task_list, tasks in domain.decision_task_lists.items():
if _task_list == task_list: if _task_list == task_list:
candidates += filter(lambda t: t.state == "SCHEDULED", tasks) candidates += [t for t in tasks if t.state == "SCHEDULED"]
if any(candidates): if any(candidates):
# TODO: handle task priorities (but not supported by boto for now) # TODO: handle task priorities (but not supported by boto for now)
task = min(candidates, key=lambda d: d.scheduled_at) task = min(candidates, key=lambda d: d.scheduled_at)
@ -293,7 +293,7 @@ class SWFBackend(BaseBackend):
candidates = [] candidates = []
for _task_list, tasks in domain.activity_task_lists.items(): for _task_list, tasks in domain.activity_task_lists.items():
if _task_list == task_list: if _task_list == task_list:
candidates += filter(lambda t: t.state == "SCHEDULED", tasks) candidates += [t for t in tasks if t.state == "SCHEDULED"]
if any(candidates): if any(candidates):
# TODO: handle task priorities (but not supported by boto for now) # TODO: handle task priorities (but not supported by boto for now)
task = min(candidates, key=lambda d: d.scheduled_at) task = min(candidates, key=lambda d: d.scheduled_at)

14
moto/swf/models/workflow_execution.py
View File

@ -202,17 +202,13 @@ class WorkflowExecution(object):
@property @property
def decision_tasks(self): def decision_tasks(self):
return filter( return [t for t in self.domain.decision_tasks
lambda t: t.workflow_execution == self, if t.workflow_execution == self]
self.domain.decision_tasks
)
@property @property
def activity_tasks(self): def activity_tasks(self):
return filter( return [t for t in self.domain.activity_tasks
lambda t: t.workflow_execution == self, if t.workflow_execution == self]
self.domain.activity_tasks
)
def _find_decision_task(self, task_token): def _find_decision_task(self, task_token):
for dt in self.decision_tasks: for dt in self.decision_tasks:
@ -294,7 +290,7 @@ class WorkflowExecution(object):
# check decision types mandatory attributes # check decision types mandatory attributes
# NB: the real SWF service seems to check attributes even for attributes list # NB: the real SWF service seems to check attributes even for attributes list
# that are not in line with the decisionType, so we do the same # that are not in line with the decisionType, so we do the same
attrs_to_check = filter(lambda x: x.endswith("DecisionAttributes"), dcs.keys()) attrs_to_check = [d for d in dcs.keys() if d.endswith("DecisionAttributes")]
if dcs["decisionType"] in self.KNOWN_DECISION_TYPES: if dcs["decisionType"] in self.KNOWN_DECISION_TYPES:
decision_type = dcs["decisionType"] decision_type = dcs["decisionType"]
decision_attr = "{0}DecisionAttributes".format(decapitalize(decision_type)) decision_attr = "{0}DecisionAttributes".format(decapitalize(decision_type))