CloseableScope

__init__(strategy='sequential') -> None

def __init__(self, strategy: Strategy = "sequential") -> None:
    self._strategy: Strategy = strategy
    self._finalizers: dict[object, Finalizer] = {}
    self._closed: Exit[Any, Any] | None = None
    self._parent: CloseableScope | None = None

add_finalizer(f) -> Effect[None]

def add_finalizer(
    self, f: Callable[[Exit[Any, Any]], Effect[object, Never, Any]]
) -> Effect[None]:
    """Register ``f`` to run on close, with the services in scope now.
    If the scope is already closed, ``f`` runs immediately."""

    def body(fiber: Fiber[Any, Any]) -> Effect[None]:
        ctx = fiber._context

        # As Effect's ``provideContext``: the closing fiber's services,
        # overridden by those in scope at registration.
        def finalizer(ex: Exit[Any, Any]) -> Effect[object, Never, Any]:
            return _Locally(_call_user(f, ex), lambda current: current.merge(ctx))

        if self._closed is not None:
            return finalizer(self._closed).as_void()
        self._add(object(), finalizer)
        return _Succeed(None)

    return _WithFiber(body)

close(ex) -> Effect[None]

def close(self, ex: Exit[Any, Any]) -> Effect[None]:
    """Run every finalizer, last added first (all at once in a parallel
    scope). Closing twice is a no-op."""

    # Uninterruptible before the scope is marked closed, so an interrupt
    # can't land between marking it and running its finalizers.
    return _Suspend(lambda: self.close_unsafe(ex) or _Succeed(None)).uninterruptible()

close_unsafe(ex) -> Effect[None] | None

def close_unsafe(self, ex: Exit[Any, Any]) -> Effect[None] | None:
    """Mark the scope closed now and return the effect that runs its
    finalizers, or ``None`` if it was closed already or had none. The
    caller must run the effect uninterruptibly (Effect's ``closeUnsafe``)."""
    if self._closed is not None:
        return None
    self._closed = ex
    if self._parent is not None:
        self._parent._remove(self)
    if not self._finalizers:
        return None
    entries = list(reversed(self._finalizers.items()))
    self._finalizers = {}
    run = _run_parallel if self._strategy == "parallel" else _run_sequential
    return run(entries, ex)

fork(strategy='sequential') -> Effect[CloseableScope]

def fork(self, strategy: Strategy = "sequential") -> Effect[CloseableScope]:
    """A child scope, closed when this one closes (if not before). The
    child of a closed scope starts closed."""

    return _Sync(lambda: self._fork_now(strategy))