pytilpack.fnctl

対応プラットフォーム

Unix専用のfcntlを使うため、Unix系の環境で利用する。

pytilpack.fnctl

fcntl関連。

lock(lock_path, retry_interval=0.1)

プロセス間で共有するファイルロックを取得する。

引数：

名前	タイプ	デスクリプション	デフォルト
lock_path	Path	ロックファイルのパス。	必須
retry_interval	float	ロックが使用中の場合に再試行するまでの待機時間（秒単位）。	0.1

ソースコード位置： pytilpack/fnctl.py

@contextlib.contextmanager
def lock(lock_path: pathlib.Path, retry_interval: float = 0.1) -> typing.Generator[None, None, None]:
    """プロセス間で共有するファイルロックを取得する。

    Args:
        lock_path: ロックファイルのパス。
        retry_interval: ロックが使用中の場合に再試行するまでの待機時間（秒単位）。
    """
    lock_path.parent.mkdir(parents=True, exist_ok=True)
    with lock_path.open("a+") as fp:
        while not _try_acquire(fp):
            time.sleep(retry_interval)
        try:
            yield
        finally:
            fcntl.flock(fp.fileno(), fcntl.LOCK_UN)

alock(lock_path, retry_interval=0.1) async

プロセス間で共有するファイルロックを取得する（async版）。

引数：

名前	タイプ	デスクリプション	デフォルト
lock_path	Path	ロックファイルのパス。	必須
retry_interval	float	ロックが使用中の場合に再試行するまでの待機時間（秒単位）。	0.1

ソースコード位置： pytilpack/fnctl.py

@contextlib.asynccontextmanager
async def alock(lock_path: pathlib.Path, retry_interval: float = 0.1) -> typing.AsyncGenerator[None, None]:
    """プロセス間で共有するファイルロックを取得する（async版）。

    Args:
        lock_path: ロックファイルのパス。
        retry_interval: ロックが使用中の場合に再試行するまでの待機時間（秒単位）。
    """
    lock_path.parent.mkdir(parents=True, exist_ok=True)
    with lock_path.open("a+") as fp:
        while not _try_acquire(fp):
            await asyncio.sleep(retry_interval)
        try:
            yield
        finally:
            fcntl.flock(fp.fileno(), fcntl.LOCK_UN)