( # ruff:ignore[too-many-arguments] # public constructor: one parameter per documented lock option
self,
lock_file: str | os.PathLike[str],
timeout: float = -1,
*,
blocking: bool = True,
is_singleton: bool = True,
loop: asyncio.AbstractEventLoop | None = None,
executor: futures.Executor | None = None,
)
| 60 | """ |
| 61 | |
| 62 | def __init__( # ruff:ignore[too-many-arguments] # public constructor: one parameter per documented lock option |
| 63 | self, |
| 64 | lock_file: str | os.PathLike[str], |
| 65 | timeout: float = -1, |
| 66 | *, |
| 67 | blocking: bool = True, |
| 68 | is_singleton: bool = True, |
| 69 | loop: asyncio.AbstractEventLoop | None = None, |
| 70 | executor: futures.Executor | None = None, |
| 71 | ) -> None: |
| 72 | creator_pid = os.getpid() |
| 73 | self._creator_pid = creator_pid |
| 74 | self._fork_invalidated = False |
| 75 | self._closed = False |
| 76 | _register_fork_object(self) |
| 77 | with _fork_transition(): |
| 78 | self._lock = ReadWriteLock(lock_file, timeout, blocking=blocking, is_singleton=is_singleton) |
| 79 | self._loop = loop |
| 80 | self._owns_executor = executor is None |
| 81 | self._executor = executor or ThreadPoolExecutor(max_workers=1) |
| 82 | if os.getpid() != creator_pid: # pragma: forked child |
| 83 | msg = "AsyncReadWriteLock construction cannot continue after fork" |
| 84 | raise RuntimeError(msg) |
| 85 | |
| 86 | @property |
| 87 | def lock_file(self) -> str: |
nothing calls this directly
no test coverage detected