enh: dyn.asdict live view, now done properly

Technologicat · Technologicat · commit f6c46c578970 · 2019-02-14T15:23:25.000+02:00
diff --git a/unpythonic/dynassign.py b/unpythonic/dynassign.py
@@ -31,15 +31,30 @@ def _getstack():
         _L._stack = _L.default_stack.copy()  # copy main thread's current stack
     return _L._stack
 
+_observers = {}
 class _EnvBlock(object):
     def __init__(self, bindings):
         self.bindings = bindings
     def __enter__(self):
         if self.bindings:  # optimization, skip pushing an empty scope
             _getstack().append(self.bindings)
+            for o in _observers.values():
+                o._refresh()
     def __exit__(self, t, v, tb):
         if self.bindings:
             _getstack().pop()
+            for o in _observers.values():
+                o._refresh()
+
+class _DynLiveView(ChainMap):
+    def __init__(self):
+        super().__init__(self)
+        self._refresh()
+        _observers[id(self)] = self
+    def __del__(self):
+        del _observers[id(self)]
+    def _refresh(self):
+        self.maps = list(reversed(_getstack())) + [_global_dynvars]
 
 class _Env(object):
     """This module exports a single object instance, ``dyn``, which provides
@@ -87,7 +102,7 @@ def main():
 
         main()
 
-    Based on StackOverflow answer by Jason Orendorff (2010).
+    Initial version of this was based on a StackOverflow answer by Jason Orendorff (2010).
 
     https://stackoverflow.com/questions/2001138/how-to-create-dynamical-scoped-variables-in-python
     """
@@ -122,21 +137,21 @@ def __contains__(self, name):
 
     # iteration
     def asdict(self):
-        """Return a view as a collections.ChainMap.
+        """Return a view of dyn as a ``collections.ChainMap``.
 
-        Note it will not detect any new scopes (or when old ones exit), because
-        the list of mappings the view uses is baked in when ``asdict()`` is called.
+        When new dynamic scopes begin or old ones exit, its ``.maps`` attribute
+        is automatically updated to reflect the changes.
         """
-        return ChainMap(*(list(reversed(_getstack())) + [_global_dynvars]))
-
-    def __iter__(self):
-        return iter(self.asdict())
-    # no __next__, iterating over dict.
+        return _DynLiveView()
 
     def items(self):
         """Abbreviation for asdict().items()."""
         return self.asdict().items()
 
+    def __iter__(self):
+        return iter(self.asdict())
+    # no __next__, iterating over dict.
+
     def __len__(self):
         return len(self.asdict())
 
diff --git a/unpythonic/test/test_dynassign.py b/unpythonic/test/test_dynassign.py
@@ -54,32 +54,41 @@ def threadtest(q):
             assert False
     runtest()
 
+    # various parts of unpythonic use dynvars, so get what's there before we insert anything for testing
+    implicits = [k for k in dyn]
+    def noimplicits(kvs):
+        return tuple(sorted((k, v) for k, v in kvs if k not in implicits))
     D = {"a": 1, "b": 2}
-    # various parts of unpythonic use dynvars, so get what's there before we insert anything
-    implicits = [k for k, v in dyn.items()]  # TODO: add a .keys() method?
-    def noimplicits(dic):
-        return {k: dic[k] for k in dic if k not in implicits}
     with dyn.let(**D):
         # membership test
         assert "a" in dyn
         assert "c" not in dyn
-        # subscript syntax as an alternative way to refer to items
+
+        # subscript syntax as an alternative notation to refer to dynamic vars
         assert dyn.a is dyn["a"]
 
-        # iteration works like in a dictionary
-        assert tuple(sorted(k for k in noimplicits(dyn))) == ("a", "b")
+        assert noimplicits(dyn.items()) == (("a", 1), ("b", 2))
+    assert noimplicits(dyn.items()) == ()
 
-        # items() gives a snapshot, with values read at the time it was called
-        assert tuple(sorted(noimplicits(dyn).items())) == (("a", 1), ("b", 2))
+    make_dynvar(im_always_there=True)
+    with dyn.let(a=1, b=2):
+        assert noimplicits(dyn.items()) == (("a", 1), ("b", 2),
+                                            ("im_always_there", True))
+    assert noimplicits(dyn.items()) == (("im_always_there", True),)
 
-        # safer (TOCTTOU) in complex situations to iterate over keys and retrieve the current dyn[k]
-        assert tuple(sorted({k: dyn[k] for k in noimplicits(dyn)}.items())) == (("a", 1), ("b", 2))
+    # dyn.asdict() returns a live view, which is essentially a collections.ChainMap
+    view = dyn.asdict()
+    assert noimplicits(view.items()) == (("im_always_there", True),)
+    with dyn.let(a=1, b=2):
+        assert noimplicits(view.items()) == (("a", 1), ("b", 2),
+                                             ("im_always_there", True))
 
-    make_dynvar(im_always_there=True)
+    # as does dyn.items() (it's an abbreviation for dyn.asdict().items())
+    items = dyn.items()
+    assert noimplicits(items) == (("im_always_there", True),)
     with dyn.let(a=1, b=2):
-        assert tuple(sorted(noimplicits(dyn).items())) == (("a", 1), ("b", 2),
-                                                           ("im_always_there", True))
-    assert tuple(sorted(noimplicits(dyn).items())) == (("im_always_there", True),)
+        assert noimplicits(items) == (("a", 1), ("b", 2),
+                                      ("im_always_there", True))
 
     print("All tests PASSED")
 
