You can not select more than 25 topics Topics must start with a chinese character,a letter or number, can include dashes ('-') and can be up to 35 characters long.

run_context.py 680 B

1234567891011121314151617181920212223
  1. from contextlib import contextmanager
  2. from contextvars import ContextVar
  3. from typing import Any, ClassVar, Generator
  4. class RunContext:
  5. RUN_CONTEXT_VAR: ClassVar[ContextVar] = ContextVar("RUN_CONTEXT_VAR")
  6. @classmethod
  7. @contextmanager
  8. def populate_context(cls, run_id) -> Generator[None, Any, None]:
  9. token = RunContext.RUN_CONTEXT_VAR.set(run_id)
  10. try:
  11. yield
  12. finally:
  13. RunContext.RUN_CONTEXT_VAR.reset(token)
  14. @classmethod
  15. def current_run_id(cls) -> str:
  16. try:
  17. return cls.RUN_CONTEXT_VAR.get()
  18. except LookupError as e:
  19. raise RuntimeError("Error getting run id") from e