commit: 47091ae95bee57f95c877b54f2eec6ea4d565c20
Author: Brian Harring <ferringb <AT> gmail <DOT> com>
AuthorDate: Mon Oct 27 06:23:27 2025 +0000
Commit: Arthur Zamarin <arthurzam <AT> gentoo <DOT> org>
CommitDate: Mon Oct 27 21:38:13 2025 +0000
URL:
https://gitweb.gentoo.org/proj/pkgcore/snakeoil.git/commit/?id=47091ae9
feat: rewrite Immutable* to drop metaclasses
The only reason this was metaclasses- and the inject form- was to work
around py2k. The metaclasses are no longer necessary, thus use this
form that just is simple inheritance, and does the nasty as needed.
Signed-off-by: Brian Harring <ferringb <AT> gmail.com>
src/snakeoil/chksum/__init__.py | 13 +-
src/snakeoil/klass/__init__.py | 3 +-
src/snakeoil/klass/immutable.py | 114 ++++++++++++++++++
src/snakeoil/klass/meta.py | 153 ------------------------
src/snakeoil/klass/util.py | 29 ++++-
tests/klass/{test_meta.py => test_immutable.py} | 78 +++++++-----
tests/klass/test_init.py | 2 -
tests/klass/test_util.py | 38 +++++-
8 files changed, 230 insertions(+), 200 deletions(-)
diff --git a/src/snakeoil/chksum/__init__.py b/src/snakeoil/chksum/__init__.py
index bc6bef2..3016b9f 100644
--- a/src/snakeoil/chksum/__init__.py
+++ b/src/snakeoil/chksum/__init__.py
@@ -6,7 +6,9 @@ import os
import sys
from importlib import import_module
-from .. import klass, osutils
+from snakeoil.klass.immutable import Simple
+
+from .. import osutils
from .defaults import chksum_loop_over_file
chksum_types = {}
@@ -130,7 +132,7 @@ def get_chksums(location, *chksums, **kwds):
)
-class LazilyHashedPath(metaclass=klass.immutable_instance):
+class LazilyHashedPath(Simple):
"""Given a pathway, compute chksums on demand via attribute access."""
def __init__(self, path, **initial_values):
@@ -139,6 +141,7 @@ class LazilyHashedPath(metaclass=klass.immutable_instance):
for attr, val in initial_values.items():
f(self, attr, val)
+ @Simple.__allow_mutation_wrapper__
def __getattr__(self, attr):
if not attr.islower():
# Disallow sHa1.
@@ -153,6 +156,7 @@ class LazilyHashedPath(metaclass=klass.immutable_instance):
object.__setattr__(self, attr, val)
return val
+ @Simple.__allow_mutation_wrapper__
def clear(self):
for key in get_handlers():
if hasattr(self, key):
@@ -161,6 +165,5 @@ class LazilyHashedPath(metaclass=klass.immutable_instance):
def __getstate__(self):
return self.__dict__.copy()
- def __setstate__(self, state):
- for k, v in state.items():
- object.__setattr__(self, k, v)
+ def __setstate__(self, data):
+ self.__dict__.update(data)
diff --git a/src/snakeoil/klass/__init__.py b/src/snakeoil/klass/__init__.py
index ca9ae9f..7f7db4b 100644
--- a/src/snakeoil/klass/__init__.py
+++ b/src/snakeoil/klass/__init__.py
@@ -46,7 +46,6 @@ from snakeoil.deprecation import deprecated as warn_deprecated
from ..caching import WeakInstMeta
from .deprecated import ImmutableInstance, immutable_instance,
inject_immutable_instance
-from .meta import combine_classes
from .properties import (
_uncached_singleton, # noqa: F401 . This exists purely due to a stupid
usage of pkgcore.ebuild.profile which is being removed.
alias,
@@ -60,7 +59,7 @@ from .properties import (
jit_attr_named,
jit_attr_none,
)
-from .util import get_attrs_of, get_slots_of
+from .util import combine_classes, get_attrs_of, get_slots_of
sentinel = object()
diff --git a/src/snakeoil/klass/immutable.py b/src/snakeoil/klass/immutable.py
new file mode 100644
index 0000000..986897c
--- /dev/null
+++ b/src/snakeoil/klass/immutable.py
@@ -0,0 +1,114 @@
+__all__ = ("Simple", "Strict")
+
+import functools
+from contextlib import contextmanager
+from contextvars import ContextVar
+
+_immutable_allow_mutations = ContextVar(
+ "immutable_instance_allow_mutation",
+ # no object's pointer will ever be zero, so this is safe.
+ default=0,
+)
+
+
+class Simple:
+ """
+ Make instance immutable, but allow __init__ to mutate.
+
+ Like :class:Strict, these protections can be sidestepped by
+ object.__setatttr__ directly. Additionally for any code decoratored with
+ :meth:allow_mutation, mutation is allowed in that invocation.
+
+ Once in a mutation block, anything that block calls is still allowed to
+ mutate this instance unless it enters another mutation block (another
instance),
+ and that block tries to mutate the original instance. TL;dr: you'll know
+ if you hit that edgecase.
+
+ >>> from snakeoil.klass.immutable import Simple
+ >>> class foo(Simple):
+ ... def __init__(self):
+ ... self.x = 1 # works
+ ... self._subinit() # works, we're in a mutation context
+ ... def _subinit(self):
+ ... # this only works if invoke in a mutation context.
+ ... # IE, __init__ for example.
+ ... self.x = 2
+ >>>
+ >>> try: foo().x =1 # doesn't
+ ... except AttributeError: pass
+ >>>
+ >>> try: foo()._subinit() # same thing, this is disallowed
+ ... except AttributError: pass
+
+ This is async and thread safe. It is not safe within generator context
+ due to a python limitation.
+ """
+
+ __slots__ = ()
+ __immutable_methods_to_autowrap__ = (
+ "__init__",
+ "__setstate__",
+ # Note, due to the mecahnism relying on id(self), the decorator
__del__ can't-
+ # even during exception an exception of the mutable block- pin the
reference
+ # forcing it to stay alive.
+ "__del__",
+ )
+
+ @contextmanager
+ def __allow_mutation__(self):
+ """Allow temporary mutation via context manager"""
+ last = _immutable_allow_mutations.set(id(self))
+ try:
+ yield
+ finally:
+ _immutable_allow_mutations.reset(last)
+
+ @classmethod
+ def __allow_mutation_wrapper__(cls, functor):
+ @functools.wraps(functor)
+ def f(instance, *args, **kwargs):
+ with cls.__allow_mutation__(instance):
+ return functor(instance, *args, **kwargs)
+
+ f.__disable_mutation_autowrapping__ = True # pyright:
ignore[reportAttributeAccessIssue] # it's already wrapped.
+ return f
+
+ def __init_subclass__(cls, *args, **kwargs) -> None:
+ """Modify the subclass allowing mutation for default allowed methods"""
+ for name in cls.__immutable_methods_to_autowrap__:
+ if (method := getattr(cls, name, None)) is not None:
+ # is it wrapped already or was marked to disable wrapping?
+ if not getattr(method, "__disable_mutation_autowrapping__",
False):
+ setattr(cls, name, cls.__allow_mutation_wrapper__(method))
+ return super().__init_subclass__(*args, **kwargs)
+
+ def __setattr__(self, name, value):
+ if id(self) != _immutable_allow_mutations.get():
+ raise AttributeError(self, name, "object is locked against
mutation")
+ object.__setattr__(self, name, value)
+
+ def __delattr__(self, attr: str):
+ if id(self) != _immutable_allow_mutations.get():
+ raise AttributeError(self, attr, "object is locked against
mutation")
+ object.__delattr__(self, attr)
+
+
+class Strict:
+ """
+ Make instances effectively immutable.
+
+ This is the 'strict' implementation; __setattr__ and __delattr__
+ will never allow mutation. Any mutation- during __init__ for example,
+ must be done via object.__setattr__(self, 'attr', value).
+
+ It's strongly advised you look at :class:Simple since that relaxes
+ the rules for things like __init__.
+ """
+
+ __slots__ = ()
+
+ def __setattr__(self, attr, _value):
+ raise AttributeError(self, attr)
+
+ def __delattr__(self, attr):
+ raise AttributeError(self, attr)
diff --git a/src/snakeoil/klass/meta.py b/src/snakeoil/klass/meta.py
deleted file mode 100644
index 899752f..0000000
--- a/src/snakeoil/klass/meta.py
+++ /dev/null
@@ -1,153 +0,0 @@
-__all__ = ("Immutable", "ImmutableStrict")
-
-import functools
-from contextlib import contextmanager
-from contextvars import ContextVar
-
-_immutable_allow_mutations = ContextVar(
- "immutable_instance_allow_mutation",
- # no object's pointer will ever be zero, so this is safe.
- default=0,
-)
-
-
-class Immutable(type):
- """
- Make instance immutable, but allow __init__ to mutate.
-
- Like :class:ImmutableStrict, these protections can be sidestepped by
- object.__setatttr__ directly. Additionally for any code decoratored with
- :meth:allow_mutation, mutation is allowed in that invocation.
-
- Once in a mutation block, anything that block calls is still allowed to
- mutate this instance unless it enters another mutation block (another
instance),
- and that block tries to mutate the original instance. TL;dr: you'll know
- if you hit that edgecase.
-
- >>> from snakeoil.klass.meta import Immutable
- >>> class foo(metaclass=Immutable):
- ... def __init__(self):
- ... self.x = 1 # works
- ... self._subinit() # works, we're in a mutation context
- ... def _subinit(self):
- ... # this only works if invoke in a mutation context.
- ... # IE, __init__ for example.
- ... self.x = 2
- >>>
- >>> try: foo().x =1 # doesn't
- ... except AttributeError: pass
- >>>
- >>> try: foo()._subinit() # same thing, this is disallowed
- ... except AttributError: pass
-
- This is async and thread safe. It is not safe within generator context
- due to a python limitation.
- """
-
- default_methods_to_wrap = (
- "__init__",
- "__setstate__",
- # Note, due to the mecahnism relying on id(self), the decorator
__del__ can't-
- # even during exception an exception of the mutable block- pin the
reference
- # forcing it to stay alive.
- "__del__",
- )
-
- class Mixin:
- # ensure that if we're in a pure slotted inheritance, we don't break
it.
- __slots__ = ()
-
- @contextmanager
- def __allow_mutation_block__(self):
- """Allow temporary mutation via context manager"""
- last = _immutable_allow_mutations.set(id(self))
- try:
- yield
- finally:
- _immutable_allow_mutations.reset(last)
-
- def __setattr__(self, name, value):
- if id(self) != _immutable_allow_mutations.get():
- raise AttributeError(self, name, "object is locked against
mutation")
- object.__setattr__(self, name, value)
-
- def __delattr__(self, attr: str):
- if id(self) != _immutable_allow_mutations.get():
- raise AttributeError(self, attr, "object is locked against
mutation")
- object.__delattr__(self, attr)
-
- @functools.wraps(type.__new__)
- def __new__(cls, name, bases, scope) -> type:
- for method in cls.default_methods_to_wrap:
- if f := scope.get(method):
- scope[method] = cls.allow_mutation(f)
-
- if not any(cls.Mixin in base.mro() for base in bases):
- bases = (cls.Mixin,) + bases
-
- return super().__new__(cls, name, bases, scope)
-
- @classmethod
- def allow_mutation(cls, functor):
- """Decorator allowing temporary mutation of an immutable instance"""
-
- @functools.wraps(functor)
- def f(self, *args, **kwargs):
- with cls.Mixin.__allow_mutation_block__(self):
- return functor(self, *args, **kwargs)
-
- return f
-
-
-class ImmutableStrict(type):
- """
- Make instances effectively immutable.
-
- This is the 'strict' implementation; __setattr__ and __delattr__
- will never allow mutation. Any mutation- during __init__ for example,
- must be done via object.__setattr__(self, 'attr', value).
-
- It's strongly advised you look at :class:Simple since that relaxes
- the rules for things like __init__.
- """
-
- class Mixin:
- __slots__ = ()
-
- def __setattr__(self, attr, _value):
- raise AttributeError(self, attr)
-
- def __delattr__(self, attr):
- raise AttributeError(self, attr)
-
- @functools.wraps(type.__new__)
- def __new__(cls, name, bases, scope):
- if not any(cls.Mixin in base.mro() for base in bases):
- bases = (cls.Mixin,) + bases
- return super().__new__(cls, name, bases, scope)
-
-
[email protected]_cache
-def combine_classes(kls: type, *extra: type) -> type:
- """Given a set of classes, combine this as if one had wrote the class by
hand
-
- This is primarily for composing metaclasses on the fly, like thus:
-
- Effectively:
- class foo(metaclass=combine_metaclasses(kls1, kls2, kls3)): pass
-
- is the same as if you did this:
- class mkls(kls1, kls2, kls3): pass
- class foo(metaclass=mkls): pass
- """
- klses = [kls]
- klses.extend(extra)
-
- if len(klses) == 1:
- return kls
-
- class combined(*klses):
- pass
-
- combined.__name__ = f"combined_{'_'.join(kls.__qualname__ for kls in
klses)}"
- return combined
diff --git a/src/snakeoil/klass/util.py b/src/snakeoil/klass/util.py
index 4701b8a..5fbe410 100644
--- a/src/snakeoil/klass/util.py
+++ b/src/snakeoil/klass/util.py
@@ -1,5 +1,6 @@
-__all__ = ("get_attrs_of", "get_slots_of")
+__all__ = ("get_attrs_of", "get_slots_of", "combine_classes")
import builtins
+import functools
from typing import Any, Iterable
_known_builtins = frozenset(
@@ -60,3 +61,29 @@ def get_attrs_of(
if (o := getattr(obj, slot, _sentinel)) is not _sentinel:
yield slot, o
seen.add(slot)
+
+
[email protected]_cache
+def combine_classes(kls: type, *extra: type) -> type:
+ """Given a set of classes, combine this as if one had wrote the class by
hand
+
+ This is primarily for composing metaclasses on the fly; this:
+
+ class foo(metaclass=combine_metaclasses(kls1, kls2, kls3)): pass
+
+ is the same as if you did this:
+
+ class mkls(kls1, kls2, kls3): pass
+ class foo(metaclass=mkls): pass
+ """
+ klses = [kls]
+ klses.extend(extra)
+
+ if len(klses) == 1:
+ return kls
+
+ class combined(*klses):
+ pass
+
+ combined.__name__ = f"combined_{'_'.join(kls.__qualname__ for kls in
klses)}"
+ return combined
diff --git a/tests/klass/test_meta.py b/tests/klass/test_immutable.py
similarity index 70%
rename from tests/klass/test_meta.py
rename to tests/klass/test_immutable.py
index 239838d..4abba2e 100644
--- a/tests/klass/test_meta.py
+++ b/tests/klass/test_immutable.py
@@ -3,7 +3,7 @@ from functools import partial, wraps
import pytest
-from snakeoil.klass import combine_classes, meta
+from snakeoil.klass import immutable
def inject_context_protection(name: str, bases: tuple[type, ...], scope) ->
type:
@@ -49,7 +49,7 @@ class
TestInjectContextProtection(metaclass=inject_context_protection):
class TestSimpleImmutable(metaclass=inject_context_protection):
- class _immutable_test_kls(metaclass=meta.Immutable):
+ class _immutable_test_kls(immutable.Simple):
def __init__(self, recurse=False):
self.dar = 1
if recurse:
@@ -58,7 +58,7 @@ class
TestSimpleImmutable(metaclass=inject_context_protection):
pytest.raises(AttributeError, setattr, o, "dar", 4)
self.dar = 3
- @meta.Immutable.allow_mutation
+ @immutable.Simple.__allow_mutation_wrapper__
def set_dar(self, value: int) -> None:
self.dar = value
@@ -69,9 +69,10 @@ class
TestSimpleImmutable(metaclass=inject_context_protection):
def setstate(self, data):
self.x = data
- class foo(metaclass=meta.Immutable):
+ class foo(immutable.Simple):
__init__ = init
+ assert foo.__init__.__disable_mutation_autowrapping__ # pyright:
ignore[reportFunctionMemberAccess]
assert foo.__init__ is not init
class foo2(foo):
@@ -80,10 +81,25 @@ class
TestSimpleImmutable(metaclass=inject_context_protection):
assert foo.__init__ is foo2.__init__
assert foo2.__setstate__ is not setstate
- # ensure that we're not daftly injecting extra instances of the
default class logic.
- # this is required to ensure we're not overriding things further down
mro.
- assert len([x for x in foo.mro() if x == meta.Immutable.Mixin]) == 1
- assert len([x for x in foo2.mro() if x == meta.Immutable.Mixin]) == 1
+ def self_mutation_managing_init(self):
+ pass
+
+ self_mutation_managing_init.__disable_mutation_autowrapping__ = True
# pyright: ignore[reportFunctionMemberAccess]
+
+ class foo3(foo2):
+ __init__ = self_mutation_managing_init
+
+ assert foo3.__init__ is self_mutation_managing_init, (
+ "__init__ was marked to not be wrapped, but got wrapped anyways"
+ )
+
+ def test_disallowed_mutation(self):
+ class kls(immutable.Simple):
+ pass
+
+ obj = kls()
+ pytest.raises(AttributeError, setattr, obj, "x", 1)
+ pytest.raises(AttributeError, delattr, obj, "y")
def test_mutation_init(self):
o = self._immutable_test_kls()
@@ -105,7 +121,7 @@ class
TestSimpleImmutable(metaclass=inject_context_protection):
assert o.dar == 1
o.set_dar(5)
assert o.dar == 5
- with o.__allow_mutation_block__():
+ with o.__allow_mutation__():
o.dar = 6
assert o.dar == 6
@@ -118,7 +134,7 @@ class
TestSimpleImmutable(metaclass=inject_context_protection):
var = contextvars.ContextVar("test", default=1)
@push_context
- def basic():
+ def basic(var=var):
assert 1 == var.get()
var.set(2)
@@ -126,7 +142,7 @@ class
TestSimpleImmutable(metaclass=inject_context_protection):
assert 1 == var.get()
@push_context
- def generator(val=2):
+ def generator(val=2, var=var):
assert 1 == var.get()
var.set(2)
yield
@@ -140,29 +156,25 @@ class
TestSimpleImmutable(metaclass=inject_context_protection):
del var
-def test_combine_metaclasses():
- class kls1(type):
- pass
-
- class kls2(type):
- pass
-
- class kls3(type):
- pass
-
- # assert it requires at least one arg
- pytest.raises(TypeError, combine_classes)
+class TestStrict:
+ def _common(self, slotted=False):
+ class kls(immutable.Strict):
+ if slotted:
+ __slots__ = ("x",)
- assert combine_classes(kls1) is kls1, "unneeded derivative metaclass was
created"
+ def m(self):
+ self.x = 1
- # assert that it refuses duplicats
- pytest.raises(TypeError, combine_classes, kls1, kls1)
+ obj = kls()
+ pytest.raises(AttributeError, setattr, obj, "x", 2)
+ pytest.raises(AttributeError, delattr, obj, "x")
+ if slotted:
+ pytest.raises(AttributeError, setattr, obj, "y", 2)
+ pytest.raises(AttributeError, delattr, obj, "y")
- # there is caching, thus also do identity check whilst checking the MRO
chain
- kls = combine_classes(kls1, kls2, kls3)
- assert kls is combine_classes(kls1, kls2, kls3), (
- "combine_metaclass uses lru_cache to avoid generating duplicate
classes, however this didn't cache"
- )
+ kls.__init__ = kls.m
+ pytest.raises(AttributeError, kls)
- combined = combine_classes(kls1, kls2)
- assert [combined, kls1, kls2, type, object] == list(combined.__mro__)
+ def test_basics(self):
+ self._common()
+ self._common(slotted=True)
diff --git a/tests/klass/test_init.py b/tests/klass/test_init.py
index bd5e6ce..92c2a52 100644
--- a/tests/klass/test_init.py
+++ b/tests/klass/test_init.py
@@ -1,4 +1,3 @@
-import contextvars
import math
import re
from functools import partial
@@ -7,7 +6,6 @@ from time import time
import pytest
from snakeoil import klass
-from snakeoil.klass import combine_classes, meta
from snakeoil.klass.properties import _internal_jit_attr, _uncached_singleton
diff --git a/tests/klass/test_util.py b/tests/klass/test_util.py
index fa13033..6039055 100644
--- a/tests/klass/test_util.py
+++ b/tests/klass/test_util.py
@@ -1,7 +1,9 @@
import weakref
from typing import Any
-from snakeoil.klass.util import get_attrs_of
+import pytest
+
+from snakeoil.klass.util import combine_classes, get_attrs_of
def test_get_attrs_of():
@@ -50,9 +52,9 @@ def test_get_attrs_of():
# the fun one. Mixed slotting.
obj = mk_obj(mk_obj(), slots="x", create=True)
obj.x = 1
- assert "x" not in obj.__dict__, (
- "a slotted variable was tucked into __dict__; this is not how python
is understood to work for this code. While real code can do this- it's dumb
but possible- this test doesn't do that, thus something is off."
- )
+ assert (
+ "x" not in obj.__dict__
+ ), "a slotted variable was tucked into __dict__; this is not how python is
understood to work for this code. While real code can do this- it's dumb but
possible- this test doesn't do that, thus something is off."
assert_attrs(obj, {"x": 1})
obj.y = 2
assert_attrs(obj, {"x": 1, "y": 2})
@@ -65,3 +67,31 @@ def test_get_attrs_of():
assert_attrs(obj, {"__weakref__": ref}, weakref=True)
obj.blah = 1
assert_attrs(obj, {}, suppressions=["blah"])
+
+
+def test_combine_classes():
+ class kls1(type):
+ pass
+
+ class kls2(type):
+ pass
+
+ class kls3(type):
+ pass
+
+ # assert it requires at least one arg
+ pytest.raises(TypeError, combine_classes)
+
+ assert combine_classes(kls1) is kls1, "unneeded derivative metaclass was
created"
+
+ # assert that it refuses duplicats
+ pytest.raises(TypeError, combine_classes, kls1, kls1)
+
+ # there is caching, thus also do identity check whilst checking the MRO
chain
+ kls = combine_classes(kls1, kls2, kls3)
+ assert (
+ kls is combine_classes(kls1, kls2, kls3)
+ ), "combine_metaclass uses lru_cache to avoid generating duplicate
classes, however this didn't cache"
+
+ combined = combine_classes(kls1, kls2)
+ assert [combined, kls1, kls2, type, object] == list(combined.__mro__)
