config.safe_writer

从 https://github.com/untitaker/python-atomicwrites 修改而来

原始版权声明
Copyright (c) 2015-2016 Markus Unterwaditzer

Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
of the Software, and to permit persons to whom the Software is furnished to do
so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

在 0.2.0 版本加入.

Attributes

FileLocks

存储文件名对应的锁

GlobalModifyLock

防止修改 FileLocks 时发生竞态条件

Classes

LockFlags

文件锁标志

SafeOpen

安全的打开文件

Functions

acquire_lock(→ None)

获取文件锁

release_lock(→ None)

释放文件锁

safe_open(…)

安全打开文件

Module Contents

class LockFlags

Bases: enum.IntEnum

文件锁标志

Initialize self. See help(type(self)) for accurate signature.

EXCLUSIVE
SHARED
class SafeOpen(io_manager: ABCTempIOManager[Any], timeout: float | None = 1, flag: LockFlags = LockFlags.EXCLUSIVE)

安全的打开文件

参数:

  • io_manager (ABCTempIOManager) -- IO管理器

  • timeout (float | None) -- 超时时间

  • flag (LockFlags) -- 锁标志

open_file(file: F) → collections.abc.Generator[F | None, Any, None]

打开文件 (上下文管理器)

参数:

file (IO) -- 文件对象

返回:

返回值为IO对象的上下文管理器

返回类型:

Generator[F | None, Any, None]

open_path(path: str | pathlib.Path, mode: str) → collections.abc.Generator[F | None, Any, None]

打开路径 (上下文管理器)

参数:
返回:

返回值为IO对象的上下文管理器

返回:

上下文管理器

返回类型:

Generator[F | None, Any, None]

_flag
_manager
_timeout = 1
acquire_lock(file: AIO, flags: LockFlags, *, timeout: numbers.Real | None = 1, immediately_release: bool = False) → None

获取文件锁

参数:

  • file (IO) -- 文件对象

  • flags (LockFlags) -- 锁类型

  • timeout (Real | None) -- 超时时间

  • immediately_release (bool) -- 是否立即释放锁

release_lock(file: AIO) → None

释放文件锁

参数:

file (IO) -- 文件对象

safe_open(path: str | pathlib.Path, mode: _typeshed.OpenBinaryMode, *, timeout: float | None = 1, flag: LockFlags = LockFlags.EXCLUSIVE, io_manager: ABCTempIOManager[Any] | None = None, **manager_kwargs: Any) → contextlib.AbstractContextManager[IO[bytes]]
safe_open(path: str | pathlib.Path, mode: _typeshed.OpenTextMode, *, timeout: float | None = 1, flag: LockFlags = LockFlags.EXCLUSIVE, io_manager: ABCTempIOManager[Any] | None = None, **manager_kwargs: Any) → contextlib.AbstractContextManager[IO[str]]

安全打开文件

参数:

  • path (str | pathlib.Path) -- 文件路径

  • mode (str) -- 打开模式

  • timeout (float | None) -- 超时时间

  • flag (LockFlags) -- 锁类型

  • io_manager (ABCTempIOManager | None) -- 临时文件管理器

  • manager_kwargs (dict) -- 临时文件管理器参数

返回:

返回IO对象的上下文管理器

返回类型:

ContextManager[IO | TextIO]

FileLocks: weakref.WeakValueDictionary[str, threading.Lock]

存储文件名对应的锁

GlobalModifyLock

防止修改 FileLocks 时发生竞态条件