Wrap code so the import request always ends up completed. If no exception raised, assume successful import. In case wrapped code needs to expose an error to the user, it should populate `import_request.status["error_messages"]` before raising an exception.
(
import_request: LaunchDarklyImportRequest,
)
| 74 | |
| 75 | @contextmanager |
| 76 | def _complete_import_request( |
| 77 | import_request: LaunchDarklyImportRequest, |
| 78 | ) -> Generator[None, None, None]: |
| 79 | """ |
| 80 | Wrap code so the import request always ends up completed. |
| 81 | |
| 82 | If no exception raised, assume successful import. |
| 83 | |
| 84 | In case wrapped code needs to expose an error to the user, it should populate |
| 85 | `import_request.status["error_messages"]` before raising an exception. |
| 86 | """ |
| 87 | result: LaunchDarklyImportResult = "incomplete" |
| 88 | try: |
| 89 | yield |
| 90 | except Exception as exc: |
| 91 | if not isinstance(exc, LaunchDarklyRateLimitError): |
| 92 | result = "failure" |
| 93 | raise exc |
| 94 | else: |
| 95 | if not import_request.status.get("error_messages"): |
| 96 | result = "success" |
| 97 | finally: |
| 98 | import_request.status["result"] = result |
| 99 | if result in {"success", "failure"}: |
| 100 | import_request.ld_token = "" |
| 101 | import_request.completed_at = timezone.now() |
| 102 | import_request.save() |
| 103 | |
| 104 | |
| 105 | def _create_environments_from_ld( |
no test coverage detected
searching dependent graphs…