(self, dag_def)
| 183 | |
| 184 | @handle_grpc_error(False) # don't retry the "RunStep" request. |
| 185 | def _run_step_impl(self, dag_def): |
| 186 | # note that the "_impl" may be retried, thus the argument cannot be a |
| 187 | # generator or an iterator. |
| 188 | runstep_requests = self._grpc_utils.generate_runstep_requests( |
| 189 | self._session_id, dag_def |
| 190 | ) |
| 191 | response = self._grpc_utils.parse_runstep_responses( |
| 192 | self._stub.RunStep(runstep_requests) |
| 193 | ) |
| 194 | |
| 195 | if response.code != coordinator_pb2.OK: |
| 196 | logger.error( |
| 197 | "Runstep failed with code: %s, message: %s", |
| 198 | coordinator_pb2.Code.Name(response.code), |
| 199 | response.error_msg, |
| 200 | ) |
| 201 | if response.full_exception: |
| 202 | exc = pickle.loads(response.full_exception) |
| 203 | if isinstance(exc, tuple): |
| 204 | raise exc[0](*exc[1:]) |
| 205 | else: |
| 206 | raise exc |
| 207 | return response |
| 208 | |
| 209 | def create_analytical_instance(self): |
| 210 | request = message_pb2.CreateAnalyticalInstanceRequest( |
no test coverage detected