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

Use dict[] to document required keys in SWF responses

Browse Source
This commit is contained in:
Jean-Baptiste Barth 2015-10-02 04:17:52 +02:00
parent 168f61c6a8
commit fbcdd5f2bd
1 changed files with 19 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
moto/swf/responses.py
View File

@ -52,8 +52,8 @@ class SWFResponse(BaseResponse):
return json.loads(self.body) return json.loads(self.body)
def _list_types(self, kind): def _list_types(self, kind):
domain_name = self._params.get("domain") domain_name = self._params["domain"]
status = self._params.get("registrationStatus") status = self._params["registrationStatus"]
reverse_order = self._params.get("reverseOrder", None) reverse_order = self._params.get("reverseOrder", None)
types = self.swf_backend.list_types(kind, domain_name, status, reverse_order=reverse_order) types = self.swf_backend.list_types(kind, domain_name, status, reverse_order=reverse_order)
return json.dumps({ return json.dumps({
@ -61,9 +61,8 @@ class SWFResponse(BaseResponse):
}) })
def _describe_type(self, kind): def _describe_type(self, kind):
domain = self._params.get("domain") domain = self._params["domain"]
_type_args = self._params.get("{}Type".format(kind)) _type_args = self._params["{}Type".format(kind)]
name = _type_args["name"] name = _type_args["name"]
version = _type_args["version"] version = _type_args["version"]
_type = self.swf_backend.describe_type(kind, domain, name, version) _type = self.swf_backend.describe_type(kind, domain, name, version)
@ -71,16 +70,16 @@ class SWFResponse(BaseResponse):
return json.dumps(_type.to_full_dict()) return json.dumps(_type.to_full_dict())
def _deprecate_type(self, kind): def _deprecate_type(self, kind):
domain = self._params.get("domain") domain = self._params["domain"]
_type = self._params.get("{}Type".format(kind)) _type_args = self._params["{}Type".format(kind)]
name = _type["name"] name = _type_args["name"]
version = _type["version"] version = _type_args["version"]
self.swf_backend.deprecate_type(kind, domain, name, version) self.swf_backend.deprecate_type(kind, domain, name, version)
return "" return ""
# TODO: implement pagination # TODO: implement pagination
def list_domains(self): def list_domains(self):
status = self._params.get("registrationStatus") status = self._params["registrationStatus"]
reverse_order = self._params.get("reverseOrder", None) reverse_order = self._params.get("reverseOrder", None)
domains = self.swf_backend.list_domains(status, reverse_order=reverse_order) domains = self.swf_backend.list_domains(status, reverse_order=reverse_order)
return json.dumps({ return json.dumps({
@ -88,20 +87,20 @@ class SWFResponse(BaseResponse):
}) })
def register_domain(self): def register_domain(self):
name = self._params.get("name") name = self._params["name"]
retention = self._params["workflowExecutionRetentionPeriodInDays"]
description = self._params.get("description") description = self._params.get("description")
retention = self._params.get("workflowExecutionRetentionPeriodInDays")
domain = self.swf_backend.register_domain(name, retention, domain = self.swf_backend.register_domain(name, retention,
description=description) description=description)
return "" return ""
def deprecate_domain(self): def deprecate_domain(self):
name = self._params.get("name") name = self._params["name"]
domain = self.swf_backend.deprecate_domain(name) domain = self.swf_backend.deprecate_domain(name)
return "" return ""
def describe_domain(self): def describe_domain(self):
name = self._params.get("name") name = self._params["name"]
domain = self.swf_backend.describe_domain(name) domain = self.swf_backend.describe_domain(name)
return json.dumps({ return json.dumps({
"configuration": { "workflowExecutionRetentionPeriodInDays": domain.retention }, "configuration": { "workflowExecutionRetentionPeriodInDays": domain.retention },
@ -113,9 +112,9 @@ class SWFResponse(BaseResponse):
return self._list_types("activity") return self._list_types("activity")
def register_activity_type(self): def register_activity_type(self):
domain = self._params.get("domain") domain = self._params["domain"]
name = self._params.get("name") name = self._params["name"]
version = self._params.get("version") version = self._params["version"]
default_task_list = self._params.get("defaultTaskList") default_task_list = self._params.get("defaultTaskList")
if default_task_list: if default_task_list:
task_list = default_task_list.get("name") task_list = default_task_list.get("name")
@ -148,9 +147,9 @@ class SWFResponse(BaseResponse):
return self._list_types("workflow") return self._list_types("workflow")
def register_workflow_type(self): def register_workflow_type(self):
domain = self._params.get("domain") domain = self._params["domain"]
name = self._params.get("name") name = self._params["name"]
version = self._params.get("version") version = self._params["version"]
default_task_list = self._params.get("defaultTaskList") default_task_list = self._params.get("defaultTaskList")
if default_task_list: if default_task_list:
task_list = default_task_list.get("name") task_list = default_task_list.get("name")