GitHub

Resources and scopes

PreviousNext

Acquire and release resources safely with acquire_release and scopes.

A scope collects finalizers and runs them when it closes. fx.acquire_release acquires a resource and registers its release on the current scope; its type then includes Scope, and fx.scoped provides a scope that closes when the effect ends.

acquire_release.py
import pyfect as fx

events: list[str] = []


class Connection:
    def query(self, sql: str) -> fx.Effect[int]:
        return fx.sync(lambda: events.append(sql)).as_(1)


# Acquire is uninterruptible; release always runs when the scope closes,
# whether the program succeeded, failed or was interrupted.
connection = fx.acquire_release(
    fx.sync(lambda: events.append("open") or Connection()),
    lambda _conn, _exit: fx.sync(lambda: events.append("close")),
)


@fx.gen
def program():
    conn = yield from connection  # needs a Scope: Effect[Connection, Never, Scope]
    return (yield from conn.query("SELECT 1"))


# scoped gives the program a scope and closes it when the program ends.
assert fx.run_sync(fx.scoped(program)) == 1
assert events == ["open", "SELECT 1", "close"]

The release runs whether the effect succeeds, fails or is interrupted, and acquisition can't be interrupted halfway. Finalizers run in reverse order of acquisition.

acquire connection, register its releaseacquire file, register its releaseclose (success, failure or interruption)release filerelease connectionuse bothprogramScopeprogramScope
acquire connection,register its releaseacquire file, register itsreleaseclose (success, failure orinterruption)release filerelease connectionuse bothprogramScopeprogramScope

Without a scope

finalizers.py
from dataclasses import dataclass

import pyfect as fx

events: list[str] = []


@dataclass
class Boom(fx.TaggedError):
    pass


# ensuring runs a finalizer however the effect ends.
failing = fx.fail(Boom()).ensuring(fx.sync(lambda: events.append("cleaned up")))
assert isinstance(fx.run_sync_exit(failing), fx.Failure)

# on_exit sees how it ended.
logged = fx.succeed(42).on_exit(lambda exit: fx.sync(lambda: events.append(repr(exit))))
assert fx.run_sync(logged) == 42

# acquire_use_release scopes a resource to one use, without a Scope.
used = fx.acquire_use_release(
    fx.succeed("file"),
    lambda f: fx.succeed(f"read {f}"),
    lambda f, _exit: fx.sync(lambda: events.append(f"closed {f}")),
)
assert fx.run_sync(used) == "read file"
assert events == ["cleaned up", "Success(value=42)", "closed file"]

In layers

A layer built with layer.effect(Tag, acquire_release(...)) owns its resource: it is acquired when the layer is built and released when the program using it ends. This is how a connection pool lives for a whole program (see examples/fastapi_style.py in the repository).

Shared and cached resources: RcRef, RcMap, ScopedRef, Resource, Pool, Cache. Reference: pyfect.scope, pyfect.pool.