ManagedRuntime

__init__(layer, *, memo_map=None) -> None

def __init__(self, layer: Layer[R_contra, ER_co], *, memo_map: MemoMap | None = None) -> None:
    """Nothing is built until the first run. Runtimes given the same
    ``memo_map`` build each layer once between them."""
    require_layer(layer, "layer")
    self.memo_map = MemoMap() if memo_map is None else memo_map
    self._layer = layer
    self._scope = CloseableScope()
    self._layer_scope = self._scope._fork_now()
    self._fiber_scope = self._scope._fork_now("parallel")
    # Fibers the runtime started (requests and the layer build), and
    # fibers running its disposal: disposal must not wait for them.
    self._managed: WeakSet[Fiber[Any, Any]] = WeakSet()
    self._disposing: WeakSet[Fiber[Any, Any]] = WeakSet()
    self._build_fiber: Fiber[Context, Any] | None = None
    self._disposed: Deferred[None] | None = None
    self._cached: Context | None = None
    self._context_effect: Effect[Context, Any] = _WithFiber(self._context_in)
    self._sync = SyncScheduler()
    self._sync_lock = threading.RLock()
    self._loop: tuple[asyncio.AbstractEventLoop, AsyncioScheduler] | None = None

make(layer, *, memo_map=None) -> ManagedRuntime[R, ER]

@staticmethod
def make[R, ER = Never](
    layer: Layer[R, ER], *, memo_map: MemoMap | None = None
) -> ManagedRuntime[R, ER]:
    return ManagedRuntime(layer, memo_map=memo_map)

context() -> Context[R_contra]

async def context(self) -> Context[R_contra]:
    """The layer's services, building it on the running loop if needed."""
    if self._cached is not None:
        return self._cached
    return unwrap(await self._run_async_unmanaged(self._context_effect))

run_fork(eff, *, scheduler=None, uninterruptible=False) -> Fiber[A, E | ER_co]

def run_fork[A, E = Never](
    self,
    eff: Effect[A, E, R_contra],
    *,
    scheduler: Scheduler | None = None,
    uninterruptible: bool = False,
) -> Fiber[A, E | ER_co]:
    """Start ``eff`` with the layer's services and return its fiber
    (see the module docstring for where it runs)."""
    require_effect(eff, "eff")
    current = current_run_scheduler()
    if current is not None:
        run: Scheduler = current
    elif _loop_running():
        run = self._loop_scheduler()
    else:
        # Another thread may be driving the runtime's scheduler: queue it.
        return self._start(
            self._sync, eff, scheduler=scheduler, uninterruptible=uninterruptible
        )
    return self._start(run, eff, scheduler=scheduler, uninterruptible=uninterruptible, now=True)

run_callback(eff, *, on_exit=None, scheduler=None, uninterruptible=False) -> Interruptor

def run_callback[A, E = Never](
    self,
    eff: Effect[A, E, R_contra],
    *,
    on_exit: Callable[[Exit[A, E | ER_co]], None] | None = None,
    scheduler: Scheduler | None = None,
    uninterruptible: bool = False,
) -> Interruptor:
    """``run_fork``, calling ``on_exit`` with the exit. Returns a
    function that interrupts the fiber."""
    require_effect(eff, "eff")
    fiber = self.run_fork(eff, scheduler=scheduler, uninterruptible=uninterruptible)
    if on_exit is not None:
        fiber.unsafe_add_observer(on_exit)

    def interrupt(by: int | None = None, /) -> None:
        fiber.unsafe_interrupt_as(by)

    return interrupt

run_sync_exit(eff) -> Exit[A, E | ER_co]

def run_sync_exit[A, E = Never](self, eff: Effect[A, E, R_contra]) -> Exit[A, E | ER_co]:
    require_effect(eff, "eff")
    # Called from one of the runtime's own sync fibers, driving its
    # scheduler would stall that fiber's run: the nested run gets its own.
    scheduler = SyncScheduler() if current_run_scheduler() is self._sync else self._sync
    fiber = self._start(scheduler, eff)
    with self._sync_lock:
        return drive(scheduler, fiber)

run_sync(eff) -> A

def run_sync[A](self: ManagedRuntime[R_contra, Never], eff: Effect[A, Never, R_contra]) -> A:
    require_effect(eff, "eff")
    return unwrap(self.run_sync_exit(eff))

run_async_exit(eff) -> Exit[A, E | ER_co]

async def run_async_exit[A, E = Never](self, eff: Effect[A, E, R_contra]) -> Exit[A, E | ER_co]:
    """Run on the running loop. Cancelling the awaiting task interrupts
    the effect and waits for its finalizers."""
    require_effect(eff, "eff")
    return await await_fiber(self._start(self._loop_scheduler(), eff))

run_async(eff) -> A

async def run_async[A](
    self: ManagedRuntime[R_contra, Never], eff: Effect[A, Never, R_contra]
) -> A:
    require_effect(eff, "eff")
    return unwrap(await self.run_async_exit(eff))

dispose() -> None

def dispose(self) -> None:
    """Dispose, blocking until the layer's resources are released.
    Raises ``RuntimeError`` inside a fiber or a running asyncio loop,
    whose fibers it could never see finish."""
    _refuse_to_block("ManagedRuntime.dispose()", "await dispose_async()")
    fiber = new_root(self._sync, Context.empty())
    fiber._start(self.dispose_effect)
    with self._sync_lock:
        unwrap(drive(self._sync, fiber))

dispose_async() -> None

async def dispose_async(self) -> None:
    """Dispose on the running loop."""
    unwrap(await self._run_async_unmanaged(self.dispose_effect))