_event_context_manager.py 1.1 KB

12345678910111213141516171819202122232425262728293031
  1. # Copyright 2021-2025 Avaiga Private Limited
  2. #
  3. # Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
  4. # the License. You may obtain a copy of the License at
  5. #
  6. # http://www.apache.org/licenses/LICENSE-2.0
  7. #
  8. # Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
  9. # an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
  10. # specific language governing permissions and limitations under the License.
  11. import typing as t
  12. from threading import Thread
  13. class _EventManager:
  14. def __init__(self) -> None:
  15. self.__thread_stack: t.List[Thread] = []
  16. def __enter__(self):
  17. return self
  18. def __exit__(self, exc_type, exc_value, traceback):
  19. if self.__thread_stack:
  20. self.__thread_stack.pop().start()
  21. if exc_value:
  22. raise exc_value
  23. return self
  24. def _add_thread(self, thread: Thread):
  25. self.__thread_stack.append(thread)