On this page

Managed spell leases๏ƒ

๐ŸŸก Intermediate ยท Lesson 46

Let a with block give the object back - ManagedSpellLease. conduit.managed_spell_meld(Spell) melds a managed spell and returns a lease: the with target is the leased object, and leaving the block releases it, even when the block raised (the block's exception is never swallowed). A lease releases exactly once: release() a second time does nothing, and lease.object raises once the object went back, because the pool may already have leased it again.

ReleaseManagedSpell().managed(obj) wraps an object you already hold - one meld returned, or one injected into your constructor - in the same kind of lease. managed_spell_meld refuses a spell that is not managed, before melding anything.

Before you run๏ƒ

Use the Intermediate guide for prerequisite concepts. Run from a checkout with Melder installed and Python 3.14 free-threading selected. The collection download includes the level's local helper modules.

Run the saved script๏ƒ

python UX_and_AIX_experiences/02_intermediate/46_managed_spell_leases.py
py -3.14t UX_and_AIX_experiences/02_intermediate/46_managed_spell_leases.py

Download this collection ยท Source on GitHub

Public surface๏ƒ

md.Conduit.managed_spell_meld, md.ManagedSpellLease (object, release, released, with), md.ReleaseManagedSpell (managed, describe), md.Existence.managed_spell_quantity, md.Existence.many, md.Conduit.meld, md.Conduit.cleanup

Code๏ƒ

  1"""
  2TIER: intermediate (46)
  3GOAL: Let a `with` block give the object back - ManagedSpellLease.
  4      conduit.managed_spell_meld(Spell) melds a managed spell and returns a lease: the
  5      `with` target is the leased object, and leaving the block releases it, even when
  6      the block raised (the block's exception is never swallowed). A lease releases
  7      exactly once: release() a second time does nothing, and lease.object raises once
  8      the object went back, because the pool may already have leased it again.
  9
 10      ReleaseManagedSpell().managed(obj) wraps an object you already hold - one meld
 11      returned, or one injected into your constructor - in the same kind of lease.
 12      managed_spell_meld refuses a spell that is not managed, before melding anything.
 13SURFACE EXERCISED: md.Conduit.managed_spell_meld, md.ManagedSpellLease (object,
 14                   release, released, with), md.ReleaseManagedSpell (managed,
 15                   describe), md.Existence.managed_spell_quantity, md.Existence.many,
 16                   md.Conduit.meld, md.Conduit.cleanup
 17"""
 18from typing import Tuple
 19
 20import melder as md
 21
 22
 23class HttpSession:
 24    """A pooled HTTP session; `requests` counts the requests made with it."""
 25
 26    def __init__(self) -> None:
 27        """Open the session."""
 28        self.requests: int = 0
 29
 30
 31class RequestLog:
 32    """A plain `many` spell, to show managed_spell_meld's refusal."""
 33
 34    def __init__(self) -> None:
 35        """Start an empty log."""
 36        self.entries: int = 0
 37
 38
 39def pool_counts(cls: type) -> Tuple[int, int, int]:
 40    """Return (alive, idle, out) for the one pool of a managed class."""
 41    (snapshot,) = md.ReleaseManagedSpell().describe(cls)
 42    return snapshot["count"], snapshot["idle_count"], snapshot["out_count"]
 43
 44
 45def main() -> None:
 46    """Use leases from managed_spell_meld and from ReleaseManagedSpell().managed.
 47
 48    Contract:
 49        Checks that a `with` block releases on exit (also when it raises), that a
 50        lease releases once, that a released lease refuses its object, and that a
 51        non-managed spell is refused.
 52    Returns:
 53        None.
 54    """
 55    book = md.Spellbook()
 56    try:
 57        book.bind(spell=HttpSession, existence=md.Existence.managed_spell_quantity)
 58        book.bind(spell=RequestLog, existence=md.Existence.many)
 59        conduit = book.conjure()
 60        try:
 61            with conduit.managed_spell_meld(HttpSession) as session:
 62                session.requests += 1
 63                assert pool_counts(HttpSession) == (1, 0, 1)
 64                print("inside the block the session is out (alive, idle, out):", pool_counts(HttpSession))
 65            assert pool_counts(HttpSession) == (1, 1, 0)
 66            print("leaving the block gave it back:", pool_counts(HttpSession))
 67
 68            # A lease held by hand: the same idle session comes back.
 69            lease = conduit.managed_spell_meld("HttpSession")
 70            assert isinstance(lease, md.ManagedSpellLease)
 71            assert lease.object is session and not lease.released
 72            lease.release()
 73            lease.release()  # a second release does nothing
 74            assert lease.released and pool_counts(HttpSession) == (1, 1, 0)
 75            try:
 76                lease.object
 77                raise AssertionError("a released lease still handed out its object")
 78            except RuntimeError as refused:
 79                print("a released lease refuses its object:", refused)
 80
 81            # The block raises: the session still goes back, and the error is not swallowed.
 82            try:
 83                with conduit.managed_spell_meld(HttpSession) as session:
 84                    session.requests += 1
 85                    raise ConnectionError("upstream timed out")
 86            except ConnectionError as error:
 87                assert pool_counts(HttpSession) == (1, 1, 0)
 88                print("the block raised", repr(error), "and the session went back anyway")
 89
 90            # Wrap an object you already hold - here, one a plain meld returned.
 91            held: HttpSession = conduit.meld("HttpSession")
 92            with md.ReleaseManagedSpell().managed(held) as same:
 93                assert same is held and pool_counts(HttpSession) == (1, 0, 1)
 94            assert pool_counts(HttpSession) == (1, 1, 0)
 95            print("managed(obj) leased a melded session and gave it back:", same is held)
 96
 97            try:
 98                conduit.managed_spell_meld("RequestLog")
 99                raise AssertionError("managed_spell_meld leased a many spell")
100            except ValueError as refused:
101                print("managed_spell_meld refuses a spell that is not managed:", str(refused)[:60], "...")
102        finally:
103            conduit.cleanup()
104    finally:
105        book.cleanup()
106
107
108if __name__ == "__main__":
109    main()

Check the outcome๏ƒ

The script contains its own assertions or demonstrated refusal paths. Run it to evaluate those checks against your installed version. The code above is taken directly from the saved file; no run output is invented here.

More intermediate examples ยท Level guide

API contracts๏ƒ