A tag for a LayerRef building the services I::
class Database(LayerRefService[Connection, DbError]): ...
DatabaseLive = Database.layer(connection_layer, idle_time_to_live=60)
Functions#
layer#
layer(layer, *, idle_time_to_live=None, preload=False, invalidation_schedule=None) -> Layer[Self, Any, Any]A layer making the LayerRef (see LayerRef.make). Provide
its dependencies with .pipe(provide(...)).
| Parameter | Type | Description |
|---|---|---|
layer | Layer[I, E, Any] | |
idle_time_to_live | Duration | None = None | |
preload | bool = False | |
invalidation_schedule | Schedule[object, object, object, Any] | None = None |
Returns Layer[Self, Any, Any]
@classmethod
def layer(
cls,
layer: Layer[I, E, Any],
*,
idle_time_to_live: Duration | None = None,
preload: bool = False,
invalidation_schedule: Schedule[object, object, object, Any] | None = None,
) -> Layer[Self, Any, Any]:
"""A layer making the ``LayerRef`` (see ``LayerRef.make``). Provide
its dependencies with ``.pipe(provide(...))``."""
require_layer(layer, "layer")
return layer_effect(
cls,
LayerRef.make(
layer,
idle_time_to_live=idle_time_to_live,
preload=preload,
invalidation_schedule=invalidation_schedule,
),
)get#
get() -> Layer[I, E, Self]A layer giving the cached services, through this tag.
Returns Layer[I, E, Self]
@classmethod
def get(cls) -> Layer[I, E, Self]:
"""A layer giving the cached services, through this tag."""
# Pyright can't solve ``unwrap``'s overloads against ``Self``.
layer_of: Effect[Any, Any, Any] = cls._service().map(lambda ref: ref.get())
return cast("Layer[I, E, Self]", unwrap(layer_of))context_effect#
context_effect() -> Effect[Context[I], E, Scope | Self]Returns Effect[Context[I], E, Scope | Self]
@classmethod
def context_effect(cls) -> Effect[Context[I], E, Scope | Self]:
return cls._service().flat_map(lambda ref: ref.context_effect())invalidate#
invalidate() -> Effect[None, Never, Self]Returns Effect[None, Never, Self]
@classmethod
def invalidate(cls) -> Effect[None, Never, Self]:
return cls._service().flat_map(lambda ref: ref.invalidate())refresh#
refresh() -> Effect[None, E, Self]Returns Effect[None, E, Self]
@classmethod
def refresh(cls) -> Effect[None, E, Self]:
return cls._service().flat_map(lambda ref: ref.refresh())