GitHub

Services and layers

PreviousNext

Declare what an effect needs; provide it with layers.

A service is a dependency an effect asks for by tag: a database, a mailer, a clock. The effect's third type parameter lists the tags it needs, and it can't run until a layer provides them.

A tag's identity is separate from its shape, so two tags (Primary, Replica) can share one interface. fx.Reference is a tag with a default value, which needs no layer.

tags.py
from typing import Protocol

import pyfect as fx
from pyfect import layer


# 1. The service's shape: any class or Protocol.
class Mailer(Protocol):
    def send(self, to: str, body: str) -> fx.Effect[None]: ...


# 2. A tag: the service's identity, which effects ask for.
class Mail(fx.Tag[Mailer]): ...


@fx.fn
def welcome(user: str):
    mailer = yield from Mail()  # needs Mail: welcome(...) is Effect[None, Never, Mail]
    yield from mailer.send(user, "Welcome!")


# 3. An implementation, provided by a layer.
class Outbox:
    def __init__(self) -> None:
        self.sent: list[str] = []

    def send(self, to: str, body: str) -> fx.Effect[None]:
        return fx.sync(lambda: self.sent.append(f"{to}: {body}"))


outbox = Outbox()
MailLive = layer.succeed(Mail, outbox)

# provide removes Mail from the requirements; run_sync accepts only Effect[A, E, Never].
fx.run_sync(welcome("ada@example.com").pipe(fx.provide(MailLive)))
assert outbox.sent == ["ada@example.com: Welcome!"]

Layers

A Layer[ROut, E, RIn] builds the services ROut, may fail with E, and needs RIn to do it.

  • Constructors: layer.succeed(Tag, service), layer.sync(Tag, f), layer.effect(Tag, eff) (which can use other services, and acquire_release for resources).
  • fx.provide(layer) on an effect, and layer.provide(other) on a layer, remove exactly what the layer builds and leave the rest of the requirements.
  • layer.merge(a, b) builds several layers concurrently.
  • Layers are memoized by identity: keep them in module-level constants, and each is built once per program however many effects use it.

For tests, layer.mock(Tag, member=...) stubs a service, and testing.layer_fixture builds a layer once per pytest scope (see Testing).

layers.py
from dataclasses import dataclass, field

import pyfect as fx
from pyfect import layer


@dataclass
class Database:
    rows: dict[str, str] = field(default_factory=lambda: {"1": "Ada"})

    def query(self, key: str) -> fx.Effect[str | None]:
        return fx.succeed(self.rows.get(key))


class Db(fx.Tag[Database]): ...


@dataclass
class UserRepo:
    db: Database

    def name(self, user_id: str) -> fx.Effect[str]:
        return self.db.query(user_id).map(lambda row: row or "unknown")


class Users(fx.Tag[UserRepo]): ...


DbLive = layer.sync(Db, Database)


# UsersLive needs Db to build: Layer[Users, Never, Db].
@fx.gen
def make_users():
    db = yield from Db()
    return UserRepo(db)


UsersLive = layer.effect(Users, make_users)

# provide on a layer satisfies its needs: Layer[Users].
AppLive = UsersLive.pipe(layer.provide(DbLive))


@fx.gen
def program():
    users = yield from Users()
    return (yield from users.name("1"))


assert fx.run_sync(program.pipe(fx.provide(AppLive))) == "Ada"

# merge builds several layers together: Layer[Db | Users].
Both = layer.merge(DbLive, AppLive)
assert fx.run_sync(program.pipe(fx.provide(Both))) == "Ada"

layer.provide(DbLive) wires DbLive's output into what UsersLive needs, so AppLive builds Db first, then Users, and needs nothing itself:

AppLive = UsersLive.pipe(layer.provide(DbLive))DbUsersDbLive builds DbUsersLive builds Usersprogram needs Users
AppLive = UsersLive.pipe(layer.provide(DbLive))DbUsersDbLive builds DbUsersLive builds Usersprogram needs Users

Reference: pyfect.layer, pyfect.context, ManagedRuntime.