SemaphoreProtocol

SyncSemaphoreProtocol

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

Bases: Protocol

Recipe for cross-process and cross-thread bounded semaphore.

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 semaphore.	required
key	Any	Key for semaphore.	required
value	int	Value for semaphore.	...
timeout	float	Timeout for semaphore.	...
expire	float | None	Expiration time for semaphore.	...
tags	str | Iterable[str] | None	Tags for semaphore.	...

Examples:

import typed_diskcache


def main() -> None:
    cache = typed_diskcache.Cache()
    semaphore = typed_diskcache.SyncSemaphore(cache, "some-key", value=2)
    semaphore.acquire()
    semaphore.acquire()
    semaphore.release()
    with semaphore:
        pass

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

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

key property

key: Any

Key for semaphore.

value property

value: int

Value for semaphore.

timeout property

timeout: float

Timeout for semaphore.

expire property

expire: float | None

Expiration time for semaphore.

tags property

tags: frozenset[str]

Tags for semaphore.

acquire

acquire() -> None

Acquire semaphore by decrementing value using spin-lock algorithm.

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

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

release

release() -> None

Release semaphore by incrementing value.

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

def release(self) -> None:
    """Release semaphore by incrementing value."""

AsyncSemaphoreProtocol

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

Bases: Protocol

Recipe for cross-process and cross-thread bounded semaphore.

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 SyncSemaphoreProtocol.

Parameters:

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

Examples:

import typed_diskcache


async def main() -> None:
        cache = typed_diskcache.Cache()
        semaphore = typed_diskcache.AsyncSemaphore(cache, "some-key", value=2)
        await semaphore.acquire()
        await semaphore.acquire()
        await semaphore.release()
        async with semaphore:
            pass

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

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

key property

key: Any

Key for semaphore.

value property

value: int

Value for semaphore.

timeout property

timeout: float

Timeout for semaphore.

expire property

expire: float | None

Expiration time for semaphore.

tags property

tags: frozenset[str]

Tags for semaphore.

acquire async

acquire() -> None

Acquire semaphore by decrementing value using spin-lock algorithm.

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

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

release async

release() -> None

Release semaphore by incrementing value.

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

async def release(self) -> None:
    """Release semaphore by incrementing value."""