Class Semaphore

src/trio/_sync.py:441–558

A `semaphore `__. A semaphore holds an integer value, which can be incremented by calling :meth:`release` and decremented by calling :meth:`acquire` – but the value is never allowed to drop below zero. If the value is zero, then

@final
class Semaphore(AsyncContextManagerMixin):
    """A `semaphore <https://en.wikipedia.org/wiki/Semaphore_(programming)>`__.

    A semaphore holds an integer value, which can be incremented by
    calling :meth:`release` and decremented by calling :meth:`acquire` – but
    the value is never allowed to drop below zero. If the value is zero, then
    :meth:`acquire` will block until someone calls :meth:`release`.

    If you&#x27;re looking for a :class:`Semaphore` to limit the number of tasks
    that can access some resource simultaneously, then consider using a
    :class:`CapacityLimiter` instead.

    This object&#x27;s interface is similar to, but different from, that of
    :class:`threading.Semaphore`.

    A :class:`Semaphore` object can be used as an async context manager; it
    blocks on entry but not on exit.

    Args:
      initial_value (int): A non-negative integer giving semaphore&#x27;s initial
        value.
      max_value (int or None): If given, makes this a "bounded" semaphore that
        raises an error if the value is about to exceed the given
        ``max_value``.

    """

    def __init__(self, initial_value: int, *, max_value: int | None = None) -> None:
        if not isinstance(initial_value, int):
            raise TypeError("initial_value must be an int")
        if initial_value < 0:
            raise ValueError("initial value must be >= 0")
        if max_value is not None:
            if not isinstance(max_value, int):
                raise TypeError("max_value must be None or an int")
            if max_value < initial_value:
                raise ValueError("max_values must be >= initial_value")

        # Invariants:
        # bool(self._lot) implies self._value == 0
        # (or equivalently: self._value > 0 implies not self._lot)
        self._lot = trio.lowlevel.ParkingLot()
        self._value = initial_value
        self._max_value = max_value

    def __repr__(self) -> str:
        if self._max_value is None:
            max_value_str = ""
        else:
            max_value_str = f", max_value={self._max_value}"
        return f"<trio.Semaphore({self._value}{max_value_str}) at {id(self):#x}>"

    @property
    def value(self) -> int:
        """The current value of the semaphore."""
        return self._value

    @property