(self, execution_id, user: User, execution_service: ExecutionService, result_folder,
file_storage)
| 60 | class _ScriptHandler: |
| 61 | |
| 62 | def __init__(self, execution_id, user: User, execution_service: ExecutionService, result_folder, |
| 63 | file_storage) -> None: |
| 64 | self.execution_id = execution_id |
| 65 | self.execution_service = execution_service |
| 66 | |
| 67 | self.config = self.execution_service.get_config(execution_id, user) |
| 68 | |
| 69 | self.result_files_paths = self._get_paths(execution_id, self._is_post_finish_path) |
| 70 | self.inline_image_paths = self._get_paths(execution_id, self._is_inline_image_path) |
| 71 | |
| 72 | self.prepared_files = {} |
| 73 | self.result_files = [] |
| 74 | self.inline_images = {} |
| 75 | |
| 76 | self.inline_image_listeners = [] |
| 77 | |
| 78 | if not self.result_files_paths and not self.inline_image_paths: |
| 79 | return |
| 80 | |
| 81 | self.output_stream = self.execution_service.get_anonymized_output_stream(execution_id) |
| 82 | |
| 83 | execution_owner = execution_service.get_owner(execution_id) |
| 84 | self.download_folder = file_storage.prepare_new_folder(execution_owner, result_folder) |
| 85 | LOGGER.info('Created download folder for ' + execution_owner + ': ' + self.download_folder) |
| 86 | |
| 87 | if self.result_files_paths: |
| 88 | execution_service.add_finish_listener(self._execution_finished, execution_id) |
| 89 | |
| 90 | if self.inline_image_paths: |
| 91 | self._listen_for_images() |
| 92 | |
| 93 | def add_inline_image_listener(self, callback): |
| 94 | self.inline_image_listeners.append(callback) |
nothing calls this directly
no test coverage detected