(
option: Literal["thread_local", "loop", "run_in_executor", "executor"],
loop: asyncio.AbstractEventLoop,
executor: Executor,
)
| 241 | |
| 242 | |
| 243 | def _unsupported_async_options( |
| 244 | option: Literal["thread_local", "loop", "run_in_executor", "executor"], |
| 245 | loop: asyncio.AbstractEventLoop, |
| 246 | executor: Executor, |
| 247 | ) -> _AsyncFileLockOptions: |
| 248 | if option == "thread_local": |
| 249 | return {"thread_local": True, "run_in_executor": False} |
| 250 | if option == "loop": |
| 251 | return {"loop": loop} |
| 252 | if option == "run_in_executor": |
| 253 | return {"run_in_executor": False} |
| 254 | return {"executor": executor} |
| 255 | |
| 256 | |
| 257 | def _build_with_explicit_defaults(kind: Literal["sync", "async"], lock_path: str) -> BaseFileLock: |
no outgoing calls
no test coverage detected