(cls)
| 29 | |
| 30 | @classmethod |
| 31 | def get_current_session(cls) -> "ThreadBasedSession": |
| 32 | curr = id(threading.current_thread()) |
| 33 | session = cls.thread2session.get(curr) |
| 34 | if session is None: |
| 35 | raise SessionNotFoundException("Can't find current session. " |
| 36 | "Maybe session closed or forget to use `register_thread()`.") |
| 37 | return session |
| 38 | |
| 39 | @classmethod |
| 40 | def get_current_task_id(cls): |
no test coverage detected