(self, **kwargs)
| 131 | |
| 132 | class BatchProcess: |
| 133 | def __init__(self, **kwargs): |
| 134 | |
| 135 | self.id = self.desc = self.cmd = self.args = self.log_dir = \ |
| 136 | self.stdout = self.stderr = self.stime = self.etime = \ |
| 137 | self.ecode = self.manager_obj = None |
| 138 | self.env = dict() |
| 139 | |
| 140 | if 'id' in kwargs: |
| 141 | self._retrieve_process(kwargs['id']) |
| 142 | else: |
| 143 | _cmd = kwargs['cmd'] |
| 144 | # Get system's interpreter |
| 145 | if kwargs['cmd'] == 'python': |
| 146 | _cmd = self._get_python_interpreter() |
| 147 | |
| 148 | self._create_process( |
| 149 | kwargs['desc'], _cmd, kwargs['args'] |
| 150 | ) |
| 151 | |
| 152 | if 'manager_obj' in kwargs: |
| 153 | self.manager_obj = kwargs['manager_obj'] |
| 154 | |
| 155 | def _retrieve_process(self, _id): |
| 156 | p = Process.for_user(pid=_id).first() |
nothing calls this directly
no test coverage detected