commit: 035ba56322b282d8ed66bd4fba1f6f057d3cf7e7
Author: Brian Harring <ferringb <AT> gmail <DOT> com>
AuthorDate: Sat Nov 29 18:28:02 2025 +0000
Commit: Brian Harring <ferringb <AT> gmail <DOT> com>
CommitDate: Sat Nov 29 18:32:56 2025 +0000
URL:
https://gitweb.gentoo.org/proj/pkgcore/snakeoil.git/commit/?id=035ba563
feat(delayed): rebuild demand_compile_regexp as delayed.regexp, delete
demandload
Folks wiped the usage- intentionally- but never bothered to wipe the
implementation.
Don't do that, especially with codebases this old- demandload did fairly
nasty things due to the time of creation making it not simple.
Mark anything you're gutting w/ deprecated and then follow through with
the wipe please.
Signed-off-by: Brian Harring <ferringb <AT> gmail.com>
src/snakeoil/delayed/__init__.py | 12 ++
src/snakeoil/demandload.py | 334 ++-------------------------------------
tests/test_delayed.py | 15 ++
tests/test_demandload.py | 135 ++--------------
4 files changed, 58 insertions(+), 438 deletions(-)
diff --git a/src/snakeoil/delayed/__init__.py b/src/snakeoil/delayed/__init__.py
new file mode 100644
index 0000000..3a469ca
--- /dev/null
+++ b/src/snakeoil/delayed/__init__.py
@@ -0,0 +1,12 @@
+__all__ = ("regexp",)
+
+import functools
+import re
+
+from ..obj import DelayedInstantiation
+
+
[email protected](re.compile)
+def regexp(pattern: str, flags: int = 0):
+ """Lazily compile a regexp; reify it only when it's needed"""
+ return DelayedInstantiation(re.Pattern, re.compile, pattern, flags)
diff --git a/src/snakeoil/demandload.py b/src/snakeoil/demandload.py
index b1d80ba..400200a 100644
--- a/src/snakeoil/demandload.py
+++ b/src/snakeoil/demandload.py
@@ -1,325 +1,23 @@
-"""Demand load things when used.
+__all__ = ("demand_compile_regexp",)
-This uses :py:func:`Placeholder` objects which create an actual object on
-first use and know how to replace themselves with that object, so
-there is no performance penalty after first use.
-
-This trick is *mostly* transparent, but there are a few things you
-have to be careful with:
-
- - You may not bind a second name to a placeholder object. Specifically,
- if you demandload C{bar} in module C{foo}, you may not
- C{from foo import bar} in a third module. The placeholder object
- does not "know" it gets imported, so this does not trigger the
- demandload: C{bar} in the third module is the placeholder object.
- When that placeholder gets used it replaces itself with the actual
- module in C{foo} but not in the third module.
- Because this is normally unwanted (it introduces a small
- performance hit) the placeholder object will raise an exception if
- it detects this. But if the demandload gets triggered before the
- third module is imported you do not get that exception, so you
- have to be careful not to import or otherwise pass around the
- placeholder object without triggering it.
- - You may not demandload more than one level of lookups. Specifically,
- demandload("os.path") is not allowed- this would require a "os" fake
- object in the local scope, one which would have a "path" fake object
- pushed into the os object. This is effectively not much of a limitation;
- you can instead just lazyload 'path' directly via demandload("os:path").
- - Not all operations on the placeholder object trigger demandload.
- The most common problem is that C{except ExceptionClass} does not
- work if C{ExceptionClass} is a placeholder.
- C{except module.ExceptionClass} with C{module} a placeholder does
- work. You can normally avoid this by always demandloading the module, not
- something in it. Another similar case is that C{isinstance Class} or
- C{issubclass Class} does not work for the initial call since the proper
- class hasn't replaced the placeholder until after the call. So the first
- call will always return False with subsequent calls working as expected. The
- previously mentioned workaround of demandloading the module works in this
- case as well.
-"""
-
-__all__ = ("demandload", "demand_compile_regexp")
-
-import functools
-import os
+import re
import sys
-import threading
-
-from .modules import load_any
-
-# There are some demandloaded imports below the definition of demandload.
-
-_allowed_chars = "".join(
- (x.isalnum() or x in "_.") and " " or "a" for x in map(chr, range(256))
-)
-
-
-def parse_imports(imports):
- """Parse a sequence of strings describing imports.
-
- For every input string it returns a tuple of (import, targetname).
- Examples::
-
- 'foo' -> ('foo', 'foo')
- 'foo:bar' -> ('foo.bar', 'bar')
- 'foo:bar,baz@spork' -> ('foo.bar', 'bar'), ('foo.baz', 'spork')
- 'foo@bar' -> ('foo', 'bar')
-
- Notice 'foo.bar' is not a valid input. Supporting 'foo.bar' would
- result in nested demandloaded objects- this isn't desirable for
- client code. Instead use 'foo:bar'.
-
- :type imports: sequence of C{str} objects.
- :rtype: iterable of tuples of two C{str} objects.
- """
- for s in imports:
- fromlist = s.split(":", 1)
- if len(fromlist) == 1:
- # Not a "from" import.
- if "." in s:
- raise ValueError(
- "dotted imports are disallowed; see "
- "snakeoil.demandload docstring for "
- f"details; {s!r}"
- )
- split = s.split("@", 1)
- for s in split:
- if not s.translate(_allowed_chars).isspace():
- raise ValueError(f"bad target: {s}")
- if len(split) == 2:
- yield tuple(split)
- else:
- split = split[0]
- yield split, split
- else:
- # "from" import.
- base, targets = fromlist
- if not base.translate(_allowed_chars).isspace():
- raise ValueError(f"bad target: {base}")
- for target in targets.split(","):
- split = target.split("@", 1)
- for s in split:
- if not s.translate(_allowed_chars).isspace():
- raise ValueError(f"bad target: {s}")
- yield base + "." + split[0], split[-1]
-
-
-def _protection_enabled_disabled():
- return False
-
-
-def _noisy_protection_disabled():
- return False
-
-
-def _protection_enabled_enabled():
- val = os.environ.get("SNAKEOIL_DEMANDLOAD_PROTECTION", "n").lower()
- return val in ("yes", "true", "1", "y")
-
-
-def _noisy_protection_enabled():
- val = os.environ.get("SNAKEOIL_DEMANDLOAD_WARN", "y").lower()
- return val in ("yes", "true", "1", "y")
-
-
-if "pydoc" in sys.modules or "epydoc" in sys.modules:
- _protection_enabled = _protection_enabled_disabled
- _noisy_protection = _noisy_protection_disabled
-else:
- _protection_enabled = _protection_enabled_enabled
- _noisy_protection = _noisy_protection_enabled
-
-
-class Placeholder:
- """Object that knows how to replace itself when first accessed.
-
- See the module docstring for common problems with its use.
- """
-
- @classmethod
- def load_namespace(cls, scope, name, target):
- """Object that imports modules into scope when first used.
-
- See the module docstring for common problems with its use; used by
- :py:func:`demandload`.
- """
- if not isinstance(target, str):
- raise TypeError(f"Asked to load non string namespace: {target!r}")
- return cls(scope, name, functools.partial(load_any, target))
-
- @classmethod
- def load_regex(cls, scope, name, *args, **kwargs):
- """
- Compiled Regex object that knows how to replace itself when first
accessed.
+import typing
- See the module docstring for common problems with its use; used by
- :py:func:`demand_compile_regexp`.
- """
- if not args and not kwargs:
- raise TypeError("re.compile requires at least one arg or kwargs")
- return cls(scope, name, functools.partial(re.compile, *args, **kwargs))
+from .delayed import regexp
+from .deprecation import deprecated
- def __init__(self, scope, name, load_func):
- """Initialize.
- :param scope: the scope we live in, normally the global namespace of
- the caller (C{globals()}).
- :param name: the name we have in C{scope}.
- :param load_func: a functor that when invoked with no args, returns the
- object we're demandloading.
- """
- if not callable(load_func):
- raise TypeError(f"load_func must be callable; got {load_func!r}")
- object.__setattr__(self, "_scope", scope)
- object.__setattr__(self, "_name", name)
- object.__setattr__(self, "_replacing_tids", [])
- object.__setattr__(self, "_load_func", load_func)
- object.__setattr__(self, "_loading_lock", threading.Lock())
+@deprecated("snakeoil.klass.demand_compile_regexp has moved to
snakeoil.delayed.regexp")
+def demand_compile_regexp(
+ name: str, pattern: str, flags=0, /, scope: dict[str, typing.Any] | None =
None
+) -> None:
+ """Lazily reify a re.compile.
- def _target_already_loaded(self, complain=True):
- name = object.__getattribute__(self, "_name")
- scope = object.__getattribute__(self, "_scope")
-
- # in a threaded environment, it's possible for tid1 to get the
- # placeholder from globals, python switches to tid2, which triggers
- # a full update (thus enabling this pathway), switch back to tid1,
- # which then throws the complaint.
- # this cannot be locked to address; the pull from global scope is
- # what would need locking, and that's infeasible (VM shouldn't do it
- # anyways; would kill performance)
- # if threading is enabled, we'll have the tid's of the threads that
- # triggered replacement; if the thread triggering this pathway isn't
- # one of the ones that caused replacement, silence the warning.
- # as for why we watch for the threading modules; if they're not there,
- # it's impossible for this pathway to accidentally be triggered twice-
- # meaning it is a misuse by the consuming client code.
- if complain:
- tids_to_complain_about = object.__getattribute__(self,
"_replacing_tids")
- if threading.current_thread().ident in tids_to_complain_about:
- if _protection_enabled():
- raise ValueError(f"Placeholder for {name!r} was triggered
twice")
- elif _noisy_protection():
- logging.warning(
- "Placeholder for %r was triggered multiple times in
file %r",
- name,
- scope.get("__file__", "unknown"),
- )
- return scope[name]
-
- def _get_target(self):
- """Replace ourself in C{scope} with the result of our C{_load_func}.
-
- :return: the result of calling C{_load_func}.
- """
- preloaded_func = object.__getattribute__(self,
"_target_already_loaded")
- with object.__getattribute__(self, "_loading_lock"):
- load_func = object.__getattribute__(self, "_load_func")
- if load_func is None:
- # This means that there was contention; two threads made it
into
- # _get_target. That's fine; suppress complaints, and return
the
- # preloaded value.
- result = preloaded_func(False)
- else:
- # We're the first thread to try and do the load; load the
target,
- # fix the scope, and replace this method with one that
shortcircuits
- # (and appropriately complains) the lookup.
- result = load_func()
- scope = object.__getattribute__(self, "_scope")
- name = object.__getattribute__(self, "_name")
- scope[name] = result
- # Replace this method with the fast path/preloaded one; this
- # is to ensure complaints get leveled if needed.
- object.__setattr__(self, "_get_target", preloaded_func)
- object.__setattr__(self, "_load_func", None)
-
- # note this step *has* to follow scope modification; else it
- # will go maximum depth recursion.
- tids = object.__getattribute__(self, "_replacing_tids")
- tids.append(threading.current_thread().ident)
-
- return result
-
- def _load_func(self):
- raise NotImplementedError
-
- # Various methods proxied to our replacement.
-
- def __str__(self):
- return self.__getattribute__("__str__")()
-
- def __getattribute__(self, attr):
- result = object.__getattribute__(self, "_get_target")()
- return getattr(result, attr)
-
- def __setattr__(self, attr, value):
- result = object.__getattribute__(self, "_get_target")()
- setattr(result, attr, value)
-
- def __call__(self, *args, **kwargs):
- result = object.__getattribute__(self, "_get_target")()
- return result(*args, **kwargs)
-
-
-def demandload(*imports, **kwargs):
- """Import modules into the caller's global namespace when each is first
used.
-
- Other args are strings listing module names.
- names are handled like this::
-
- foo import foo
- foo@bar import foo as bar
- foo:bar from foo import bar
- foo:bar,quux from foo import bar, quux
- foo.bar:quux from foo.bar import quux
- foo:baz@quux from foo import baz as quux
+ The mechanism of injecting into the scope is deprecated; move to
snakeoil.delayed.regexp.
"""
-
- # pull the caller's global namespace if undefined
- scope = kwargs.pop("scope", sys._getframe(1).f_globals)
-
- for source, target in parse_imports(imports):
- scope[target] = Placeholder.load_namespace(scope, target, source)
-
-
-# Extra name to make undoing monkeypatching demandload with
-# disabled_demandload easier.
-enabled_demandload = demandload
-
-
-def disabled_demandload(*imports, **kwargs):
- """Exactly like :py:func:`demandload` but does all imports immediately."""
- scope = kwargs.pop("scope", sys._getframe(1).f_globals)
- for source, target in parse_imports(imports):
- scope[target] = load_any(source)
-
-
-def demand_compile_regexp(name, *args, **kwargs):
- """Demandloaded version of :py:func:`re.compile`.
-
- Extra arguments are passed unchanged to :py:func:`re.compile`.
-
- :param name: the name of the compiled re object in that scope.
- """
- scope = kwargs.pop("scope", sys._getframe(1).f_globals)
- scope[name] = Placeholder.load_regex(scope, name, *args, **kwargs)
-
-
-def disabled_demand_compile_regexp(name, *args, **kwargs):
- """Exactly like :py:func:`demand_compile_regexp` but does all imports
immediately."""
- scope = kwargs.pop("scope", sys._getframe(1).f_globals)
- scope[name] = re.compile(*args, **kwargs)
-
-
-if os.environ.get("SNAKEOIL_DEMANDLOAD_DISABLED", "n").lower() in (
- "y",
- "yes",
- "1",
- "true",
-):
- demandload = disabled_demandload
- demand_compile_regexp = disabled_demand_compile_regexp
-
-demandload(
- "logging",
- "re",
-)
+ if scope is None:
+ # Note 2, not 1- we're wrapped in deprecated so we *are* two levels in.
+ scope = sys._getframe(2).f_globals
+ delayed = regexp(pattern, flags)
+ scope[name] = delayed
diff --git a/tests/test_delayed.py b/tests/test_delayed.py
new file mode 100644
index 0000000..c74f454
--- /dev/null
+++ b/tests/test_delayed.py
@@ -0,0 +1,15 @@
+import re
+
+from snakeoil import delayed
+
+
+def test_regexp():
+ d = delayed.regexp("aasdf", 1)
+ assert re.Pattern is not type(d), "a proxy wasn't returned"
+ assert "aasdf" == d.pattern
+ assert re.compile("asdf", 1).flags == d.flags
+ assert d.match("aasdf")
+ assert re.compile("fdas").flags == delayed.regexp("").flags
+
+ # assert we lie.
+ assert isinstance(delayed.regexp("asdf"), re.Pattern)
diff --git a/tests/test_demandload.py b/tests/test_demandload.py
index 0ef6d9a..12daf1a 100644
--- a/tests/test_demandload.py
+++ b/tests/test_demandload.py
@@ -2,127 +2,22 @@ import re
import pytest
-from snakeoil import demandload
-
-# few notes:
-# all tests need to be wrapped w/ the following decorator; it
-# ensures that snakeoils env-aware disabling is reversed, ensuring the
-# setup is what the test expects.
-# it also explicitly resets the state on the way out.
-
-
-def reset_globals(functor):
- def f(*args, **kwds):
- orig_demandload = demandload.demandload
- orig_demand_compile = demandload.demand_compile_regexp
- orig_protection = demandload._protection_enabled
- orig_noisy = demandload._noisy_protection
- try:
- return functor(*args, **kwds)
- finally:
- demandload.demandload = orig_demandload
- demandload.demand_compile_regexp = orig_demand_compile
- demandload._protection_enabled = orig_protection
- demandload._noisy_protection = orig_noisy
-
- return f
-
-
-class TestParser:
- @reset_globals
- def test_parse(self):
- for input, output in [
- ("foo", [("foo", "foo")]),
- ("foo:bar", [("foo.bar", "bar")]),
- ("foo:bar,baz@spork", [("foo.bar", "bar"), ("foo.baz", "spork")]),
- ("foo@bar", [("foo", "bar")]),
- ("foo_bar", [("foo_bar", "foo_bar")]),
- ]:
- assert output == list(demandload.parse_imports([input]))
- pytest.raises(ValueError, list, demandload.parse_imports(["a.b"]))
- pytest.raises(ValueError, list, demandload.parse_imports(["a:,"]))
- pytest.raises(ValueError, list, demandload.parse_imports(["a:b,x@"]))
- pytest.raises(ValueError, list, demandload.parse_imports(["b-x"]))
- pytest.raises(ValueError, list, demandload.parse_imports([" b_x"]))
-
-
-class TestPlaceholder:
- @reset_globals
- def test_getattr(self):
- scope = {}
- placeholder = demandload.Placeholder(scope, "foo", list)
- assert scope == object.__getattribute__(placeholder, "_scope")
- assert placeholder.__doc__ == [].__doc__
- assert scope["foo"] == []
- demandload._protection_enabled = lambda: True
- with pytest.raises(ValueError):
- getattr(placeholder, "__doc__")
-
- @reset_globals
- def test__str__(self):
- scope = {}
- placeholder = demandload.Placeholder(scope, "foo", list)
- assert scope == object.__getattribute__(placeholder, "_scope")
- assert str(placeholder) == str([])
- assert scope["foo"] == []
-
- @reset_globals
- def test_call(self):
- def passthrough(*args, **kwargs):
- return args, kwargs
-
- def get_func():
- return passthrough
-
- scope = {}
- placeholder = demandload.Placeholder(scope, "foo", get_func)
- assert scope == object.__getattribute__(placeholder, "_scope")
- assert (("arg",), {"kwarg": 42}) == placeholder("arg", kwarg=42)
- assert passthrough is scope["foo"]
-
- @reset_globals
- def test_setattr(self):
- class Struct:
- pass
-
- scope = {}
- placeholder = demandload.Placeholder(scope, "foo", Struct)
- placeholder.val = 7
- demandload._protection_enabled = lambda: True
- with pytest.raises(ValueError):
- getattr(placeholder, "val")
- assert 7 == scope["foo"].val
-
-
-class TestImport:
- @reset_globals
- def test_demandload(self):
- scope = {}
- demandload.demandload("snakeoil:demandload", scope=scope)
- assert demandload is not scope["demandload"]
- assert demandload.demandload is scope["demandload"].demandload
- assert demandload is scope["demandload"]
-
- @reset_globals
- def test_disabled_demandload(self):
- scope = {}
- demandload.disabled_demandload("snakeoil:demandload", scope=scope)
- assert demandload is scope["demandload"]
+from snakeoil import demandload, deprecation
class TestDemandCompileRegexp:
- @reset_globals
def test_demand_compile_regexp(self):
- scope = {}
- demandload.demand_compile_regexp("foo", "frob", scope=scope)
- assert list(scope.keys()) == ["foo"]
- assert "frob" == scope["foo"].pattern
- assert "frob" == scope["foo"].pattern
-
- # verify it's delayed via a bad regex.
- demandload.demand_compile_regexp("foo", "f(", scope=scope)
- assert list(scope.keys()) == ["foo"]
- # should blow up on accessing an attribute.
- obj = scope["foo"]
- with pytest.raises(re.error):
- getattr(obj, "pattern")
+ with deprecation.suppress_deprecation_warning():
+ scope = {}
+ demandload.demand_compile_regexp("foo", "frob", scope=scope)
+ assert list(scope.keys()) == ["foo"]
+ assert "frob" == scope["foo"].pattern
+ assert "frob" == scope["foo"].pattern
+
+ # verify it's delayed via a bad regex.
+ demandload.demand_compile_regexp("foo", "f(", scope=scope)
+ assert list(scope.keys()) == ["foo"]
+ # should blow up on accessing an attribute.
+ obj = scope["foo"]
+ with pytest.raises(re.error):
+ getattr(obj, "pattern")
