LockProtocol

SyncLockProtocol

SyncLockProtocol(
    cache: CacheProtocol,
    key: Any,
    *,
    timeout: float = ...,
    expire: float | None = ...,
    tags: str | Iterable[str] | None = ...
)

Bases: Protocol

Recipe for cross-process and cross-thread lock.

Assumes the key will not be evicted. Set the eviction policy to 'none' on the cache to guarantee the key is not evicted.

Parameters:

Name	Type	Description	Default
cache	CacheProtocol	Cache to use for lock.	required
key	Any	Key for lock.	required
timeout	float	Timeout for lock.	...
expire	float | None	Expiration time for lock.	...
tags	str | Iterable[str] | None	Tags for lock.	...

Examples:

import typed_diskcache


def main() -> None:
    cache = typed_diskcache.Cache()
    lock = typed_diskcache.SyncLock(cache, "some-key")
    lock.acquire()
    lock.release()
    with lock:
        pass

Source code in src/typed_diskcache/interface/sync.py

def __init__(
    self,
    cache: CacheProtocol,
    key: Any,
    *,
    timeout: float = ...,
    expire: float | None = ...,
    tags: str | Iterable[str] | None = ...,
) -> None: ...

key property

key: Any

Key for lock.

timeout property

timeout: float

Timeout for lock.

expire property

expire: float | None

Expiration time for lock.

tags property

tags: frozenset[str]

Tags for lock.

locked property

locked: bool

Return true if the lock is acquired.

acquire

acquire() -> None

Acquire lock using spin-lock algorithm.

Source code in src/typed_diskcache/interface/sync.py

def acquire(self) -> None:
    """Acquire lock using spin-lock algorithm."""

release

release() -> None

Release lock by deleting key.

Source code in src/typed_diskcache/interface/sync.py

def release(self) -> None:
    """Release lock by deleting key."""

AsyncLockProtocol

AsyncLockProtocol(
    cache: CacheProtocol,
    key: Any,
    *,
    timeout: float = ...,
    expire: float | None = ...,
    tags: str | Iterable[str] | None = ...
)

Bases: Protocol

Recipe for cross-process and cross-thread lock.

Assumes the key will not be evicted. Set the eviction policy to 'none' on the cache to guarantee the key is not evicted.

Asynchronous version of SyncLockProtocol.

Parameters:

Name	Type	Description	Default
cache	CacheProtocol	Cache to use for lock.	required
key	Any	Key for lock.	required
timeout	float	Timeout for lock.	...
expire	float | None	Expiration time for lock.	...
tags	str | Iterable[str] | None	Tags for lock.	...

Examples:

import typed_diskcache


async def main() -> None:
    cache = typed_diskcache.Cache()
    lock = typed_diskcache.AsyncLock(cache, "some-key")
    await lock.acquire()
    await lock.release()
    async with lock:
        pass

Source code in src/typed_diskcache/interface/sync.py

def __init__(
    self,
    cache: CacheProtocol,
    key: Any,
    *,
    timeout: float = ...,
    expire: float | None = ...,
    tags: str | Iterable[str] | None = ...,
) -> None: ...

key property

key: Any

Key for lock.

timeout property

timeout: float

Timeout for lock.

expire property

expire: float | None

Expiration time for lock.

tags property

tags: frozenset[str]

Tags for lock.

locked property

locked: bool

Return true if the lock is acquired.

acquire async

acquire() -> None

Acquire lock using spin-lock algorithm.

Source code in src/typed_diskcache/interface/sync.py

async def acquire(self) -> None:
    """Acquire lock using spin-lock algorithm."""

release async

release() -> None

Release lock by deleting key.

Source code in src/typed_diskcache/interface/sync.py

async def release(self) -> None:
    """Release lock by deleting key."""