(self, title, body, files=None)
| 14 | self._communicator = _create_communicator(params_dict, process_invoker) |
| 15 | |
| 16 | def send(self, title, body, files=None): |
| 17 | environment_variables = None |
| 18 | |
| 19 | if isinstance(body, dict): |
| 20 | parameters = list(body.values()) |
| 21 | environment_variables = values_to_string(dict(body)) |
| 22 | elif isinstance(body, list): |
| 23 | parameters = body |
| 24 | else: |
| 25 | raise Exception('Only dict or list bodies are supported') |
| 26 | |
| 27 | parameters = values_to_string(parameters) |
| 28 | |
| 29 | if files: |
| 30 | raise Exception('Files are not supported for scripts') |
| 31 | |
| 32 | self._communicator.send(parameters, environment_variables=environment_variables) |
| 33 | |
| 34 | def __str__(self, *args, **kwargs): |
| 35 | return type(self).__name__ + ' for ' + str(self._communicator) |
nothing calls this directly
no test coverage detected