Internal timeout context manager. Use :func:`pymongo.timeout` instead:: with pymongo.timeout(0.5): client.test.test.insert_one({})
| 69 | |
| 70 | |
| 71 | class _TimeoutContext(AbstractContextManager[Any]): |
| 72 | """Internal timeout context manager. |
| 73 | |
| 74 | Use :func:`pymongo.timeout` instead:: |
| 75 | |
| 76 | with pymongo.timeout(0.5): |
| 77 | client.test.test.insert_one({}) |
| 78 | """ |
| 79 | |
| 80 | def __init__(self, timeout: Optional[float]): |
| 81 | self._timeout = timeout |
| 82 | self._tokens: Optional[tuple[Token[Optional[float]], Token[float], Token[float]]] = None |
| 83 | |
| 84 | def __enter__(self) -> None: |
| 85 | timeout_token = TIMEOUT.set(self._timeout) |
| 86 | prev_deadline = DEADLINE.get() |
| 87 | next_deadline = time.monotonic() + self._timeout if self._timeout else float("inf") |
| 88 | deadline_token = DEADLINE.set(min(prev_deadline, next_deadline)) |
| 89 | rtt_token = RTT.set(0.0) |
| 90 | self._tokens = (timeout_token, deadline_token, rtt_token) |
| 91 | |
| 92 | def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: |
| 93 | if self._tokens: |
| 94 | timeout_token, deadline_token, rtt_token = self._tokens |
| 95 | TIMEOUT.reset(timeout_token) |
| 96 | DEADLINE.reset(deadline_token) |
| 97 | RTT.reset(rtt_token) |
| 98 | |
| 99 | |
| 100 | # See https://mypy.readthedocs.io/en/stable/generics.html?#decorator-factories |