Python allows the creation of classes to be modified via metaclasses. They can perform all sorts of operations that might appear to be magical to languages that do not have the capability. The goal of the following metaclass is to automatically add a mutex to all method calls.

It is easy to write such a metaclass if the possibility of inheritance is ignored. When base classes implemented in Python are used, their methods must also be considered, but those written in C can usually be safely ignored. This means the metaclass must be somewhat recursive.

Does the code shown below accomplish the goal of making the classes it builds thread-safe, and can the metaclass be improved for clarity?

class AtomicMeta(type):
    """AtomicMeta(name, bases, dictionary) -> new thread-safe class"""
    __REGISTRY = {}

    def __new__(mcs, name, bases, dictionary, parent=None):
        """Create a new class while fixing bases and all callable items."""
        final_bases = []
        # Replace bases with those that are safe to use.
        for base in bases:
            fixed = mcs.__REGISTRY.get(base)
            if fixed:
            elif base in mcs.__REGISTRY.values():
            elif base in vars(builtins).values():
                    base.__name__, base.__bases__, dict(vars(base)), base
        class_lock = threading.Lock()
        # Wrap all callable attributes so that they are thread-safe.
        for key, value in dictionary.items():
            if callable(value):
                dictionary[key] = mcs.__wrap(value, class_lock)
        new_class = super().__new__(mcs, name, tuple(final_bases), dictionary)
        # Register the class and potentially replace parent references.
        if parent is None:
            mcs.__REGISTRY[object()] = new_class
            mcs.__REGISTRY[parent] = new_class
            source = inspect.getmodule(parent)
            *prefix, root = parent.__qualname__.split('.')
            for name in prefix:
                source = getattr(source, name)
            setattr(source, root, new_class)
        return new_class

    # noinspection PyUnusedLocal
    def __init__(cls, name, bases, dictionary, parent=None):
        """Initialize the new class while ignoring any potential parent."""
        super().__init__(name, bases, dictionary)

    def __wrap(method, class_lock):
        """Ensure that all method calls are run as atomic operations."""

        def atomic_wrapper(self, *args, **kwargs):
            with class_lock:
                    instance_lock = self.__lock
                except AttributeError:
                    instance_lock = self.__lock = threading.RLock()
            with instance_lock:
                return method(self, *args, **kwargs)

        return atomic_wrapper
  • \$\begingroup\$ Is it intentional that your setattr(source, root, new_class) is outside of the for name in prefix loop? If so, I'd probably skip the loop and do setattr(source[-1], root, new_class) \$\endgroup\$ – C.Nivs Jul 14 '19 at 16:24
  • \$\begingroup\$ @C.Nivs __qualname__ should have the full path to an object which may have periods in the name. In order to get the actual object (source), the name has to be split on periods; and each object in the path must be retrieved sequentially. \$\endgroup\$ – Noctis Skytower Jul 15 '19 at 1:39
  • \$\begingroup\$ Ah I see now, it's just recursively building that attribute \$\endgroup\$ – C.Nivs Jul 15 '19 at 1:59

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.