Add scoped singleton registries to SingletonConfigurable

.pre-commit-config.yaml

@@ -4,7 +4,7 @@ ci:
 
 repos:
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.5.0
+    rev: v6.0.0
     hooks:
       - id: check-case-conflict
       - id: check-ast

docs/source/config-api.rst

@@ -9,6 +9,9 @@ Traitlets config API reference
 .. autoclass:: SingletonConfigurable
     :members:
 
+.. autoclass:: SingletonScope
+    :members:
+
 .. autoclass:: LoggingConfigurable
     :members:
 

docs/source/config.rst

@@ -73,6 +73,97 @@ Singletons: :class:`~traitlets.config.SingletonConfigurable`
     of a given singleton class, but the :meth:`instance` method will always
     return the same one.
 
+Singleton scopes
+----------------
+
+By default :meth:`~traitlets.config.SingletonConfigurable.instance` resolves
+against a single, process-global registry. A :class:`~traitlets.config.SingletonScope`
+provides an isolated registry that is active only for a dynamic extent (the
+current thread or :mod:`asyncio` task), leaving the global one untouched. This is
+useful when you need to run code that internally calls ``.instance()`` — code you
+do not control — against a fresh set of singletons, for example in a test or when
+serving concurrent requests.
+
+Create a scope from the base class whose subtree it should cover with
+:meth:`~traitlets.config.SingletonConfigurable.scope`, then activate it as a
+context manager:
+
+.. sourcecode:: python
+
+    a = MyApp.instance()  # process-global instance
+
+    scope = MyApp.scope()  # covers MyApp and its subclasses
+    with scope():
+        third_party_function()  # its internal MyApp.instance() calls...
+        b = MyApp.instance()  # ...resolve to this fresh instance
+        assert b is not a
+    assert MyApp.instance() is a  # global untouched, exception-safe
+
+    scope.get(MyApp)  # -> b, retrieve what was created in-scope
+    with scope():  # re-entering resumes the same registry
+        assert MyApp.instance() is b
+
+Key semantics:
+
+- **Coverage is defined by the base class.** ``Foo.scope()`` only intercepts
+  ``.instance()`` for ``Foo`` and its subclasses; unrelated singletons keep
+  resolving against the global registry. Use ``SingletonConfigurable.scope()`` to
+  cover every singleton, or a narrower base to limit the blast radius.
+- **No fallback to the global.** Inside a scope, the first ``.instance()`` call
+  creates a fresh instance registered in the scope; :meth:`instance` never reads,
+  creates, or mutates the process-global ``_instance``, and
+  :meth:`~traitlets.config.SingletonConfigurable.initialized` is ``False`` until
+  that first in-scope creation.
+- **Pre-seeding.** :meth:`~traitlets.config.SingletonScope.add` registers an
+  existing instance so it is returned inside the scope (including via ancestor
+  singleton classes), letting you reuse an object across scope entries.
+- **Nesting and propagation.** Scopes nest; the innermost active scope covering a
+  class wins, and blocks restore in LIFO order (even on exceptions). Because the
+  active scope lives in a :class:`~contextvars.ContextVar`, an :mod:`asyncio` task
+  created inside a scope inherits it, but a new :class:`threading.Thread` started
+  inside does not — it falls through to the global registry.
+- **Direct construction is never intercepted.** ``MyApp()`` always builds a new,
+  unregistered object, in or out of a scope.
+
+Because a scope is bound to the thread (or task) that activates it, a freshly
+started :class:`threading.Thread` does *not* inherit its parent's scope. This
+makes it easy to give each worker thread its own isolated singleton: have every
+thread activate its own scope. Any ``.instance()`` calls made by that thread —
+directly or deep inside code it calls — then resolve to a per-thread instance,
+while the process-global singleton is left untouched:
+
+.. sourcecode:: python
+
+    import threading
+    from traitlets.config import SingletonConfigurable
+
+
+    class MyApp(SingletonConfigurable):
+        pass
+
+
+    seen = {}
+
+
+    def worker(name):
+        scope = MyApp.scope()  # a fresh registry, private to this thread
+        with scope():  # activate it for this thread's extent
+            app = MyApp.instance()  # created once, here...
+            assert MyApp.instance() is app  # ...and reused within the thread
+            seen[name] = app
+
+
+    threads = [threading.Thread(target=worker, args=(i,)) for i in range(3)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    # each thread got its own distinct instance
+    assert len({id(app) for app in seen.values()}) == 3
+    # and none of them is the process-global singleton
+    assert all(app is not MyApp.instance() for app in seen.values())
+
 Having described these main concepts, we can now state the main idea in our
 configuration system: *"configuration" allows the default values of class
 attributes to be controlled on a class by class basis*. Thus all instances of

tests/config/test_configurable.py

@@ -4,14 +4,21 @@
 # Distributed under the terms of the Modified BSD License.
 from __future__ import annotations
 
+import asyncio
 import logging
+import threading
 from unittest import TestCase
 
 import pytest
 
 from tests._warnings import expected_warnings
 from traitlets.config.application import Application
-from traitlets.config.configurable import Configurable, LoggingConfigurable, SingletonConfigurable
+from traitlets.config.configurable import (
+    Configurable,
+    LoggingConfigurable,
+    MultipleInstanceError,
+    SingletonConfigurable,
+)
 from traitlets.config.loader import Config
 from traitlets.log import get_logger
 from traitlets.traitlets import (
@@ -335,6 +342,195 @@ class Bam(Bar):
         self.assertEqual(SingletonConfigurable._instance, None)
 
 
+class TestSingletonScope(TestCase):
+    def test_isolation(self):
+        class MyApp(SingletonConfigurable):
+            pass
+
+        a = MyApp.instance()
+        scope = MyApp.scope()
+        with scope():
+            b = MyApp.instance()
+            self.assertIsNot(b, a)
+            self.assertIs(MyApp.instance(), b)
+
+        self.assertIs(MyApp.instance(), a)
+        self.assertIs(MyApp._instance, a)
+
+    def test_lazy_creation_uncontrolled_code(self):
+        class MyApp(SingletonConfigurable):
+            pass
+
+        def third_party():
+            return MyApp.instance()
+
+        scope = MyApp.scope()
+        with scope():
+            b = third_party()
+            self.assertIs(scope.get(MyApp), b)
+
+    def test_no_global_side_effects(self):
+        class MyApp(SingletonConfigurable):
+            pass
+
+        scope = MyApp.scope()
+        with scope():
+            MyApp.instance()
+
+        self.assertIsNone(MyApp._instance)
+        self.assertIs(MyApp.initialized(), False)
+
+    def test_initialized_false_before_creation_in_scope(self):
+        class MyApp(SingletonConfigurable):
+            pass
+
+        MyApp.instance()
+        self.assertIs(MyApp.initialized(), True)
+
+        scope = MyApp.scope()
+        with scope():
+            self.assertIs(MyApp.initialized(), False)
+            MyApp.instance()
+            self.assertIs(MyApp.initialized(), True)
+
+    def test_reentry(self):
+        class MyApp(SingletonConfigurable):
+            pass
+
+        scope = MyApp.scope()
+        with scope():
+            b = MyApp.instance()
+
+        with scope():
+            self.assertIs(MyApp.instance(), b)
+
+    def test_preseeding(self):
+        class Bar(SingletonConfigurable):
+            pass
+
+        class Bam(Bar):
+            pass
+
+        existing = Bam()
+        scope = Bar.scope()
+        scope.add(existing)
+        with scope():
+            self.assertIs(Bam.instance(), existing)
+            self.assertIs(Bar.instance(), existing)
+
+    def test_blast_radius(self):
+        class Foo(SingletonConfigurable):
+            pass
+
+        class Baz(SingletonConfigurable):
+            pass
+
+        scope = Foo.scope()
+        with scope():
+            Foo.instance()
+            self.assertIsNone(Foo._instance)
+
+            baz = Baz.instance()
+            self.assertIs(Baz._instance, baz)
+            self.assertIsNone(scope.get(Baz))
+
+    def test_nesting(self):
+        class Outer(SingletonConfigurable):
+            pass
+
+        outer_scope = Outer.scope()
+        inner_scope = Outer.scope()
+
+        with outer_scope():
+            o = Outer.instance()
+
+            with inner_scope():
+                i = Outer.instance()
+                self.assertIsNot(i, o)
+
+            self.assertIs(Outer.instance(), o)
+
+            try:
+                with inner_scope():
+                    raise RuntimeError("boom")
+            except RuntimeError:
+                pass
+
+            self.assertIs(Outer.instance(), o)
+
+    def test_mro_parity_in_scope(self):
+        class Bar(SingletonConfigurable):
+            pass
+
+        class Bam(Bar):
+            pass
+
+        scope1 = Bar.scope()
+        with scope1():
+            bam = Bam.instance()
+            self.assertIs(Bar.instance(), bam)
+
+        scope2 = Bar.scope()
+        with scope2():
+            Bar.instance()
+            with self.assertRaises(MultipleInstanceError):
+                Bam.instance()
+
+    def test_clear_instance_in_scope(self):
+        class MyApp(SingletonConfigurable):
+            pass
+
+        a = MyApp.instance()
+        scope = MyApp.scope()
+        with scope():
+            b = MyApp.instance()
+            MyApp.clear_instance()
+            self.assertIsNone(scope.get(MyApp))
+            self.assertIs(MyApp._instance, a)
+
+            c = MyApp.instance()
+            self.assertIsNot(c, b)
+
+    def test_thread_isolation(self):
+        class MyApp(SingletonConfigurable):
+            pass
+
+        a = MyApp.instance()
+        scope = MyApp.scope()
+        results = {}
+        with scope():
+            b = MyApp.instance()
+
+            def target():
+                results["thread"] = MyApp.instance()
+
+            thread = threading.Thread(target=target)
+            thread.start()
+            thread.join()
+
+            self.assertIsNot(results["thread"], b)
+            self.assertIs(results["thread"], a)
+            self.assertIs(MyApp._instance, a)
+
+    def test_async_propagation(self):
+        class MyApp(SingletonConfigurable):
+            pass
+
+        scope = MyApp.scope()
+
+        async def coro():
+            return MyApp.instance()
+
+        async def main():
+            with scope():
+                b = MyApp.instance()
+                task = asyncio.create_task(coro())
+                result = await task
+                self.assertIs(result, b)
+
+        asyncio.run(main())
+
+
 class TestLoggingConfigurable(TestCase):
     def test_parent_logger(self):
         class Parent(LoggingConfigurable):

traitlets/config/__init__.py

@@ -16,5 +16,6 @@
     "LoggingConfigurable",
     "MultipleInstanceError",
     "SingletonConfigurable",
+    "SingletonScope",
     "configurable",
 ]