# Writing a metaclass to ensure a class is thread-safe

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:
final_bases.append(fixed)
elif base in mcs.__REGISTRY.values():
final_bases.append(base)
elif base in vars(builtins).values():
final_bases.append(base)
else:
final_bases.append(mcs(
base.__name__, base.__bases__, dict(vars(base)), base
))
# 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
else:
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)

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

@functools.wraps(method)
def atomic_wrapper(self, *args, **kwargs):
with class_lock:
try:
instance_lock = self.__lock
except AttributeError:

• 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) – C.Nivs Jul 14 '19 at 16:24
• @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. – Noctis Skytower Jul 15 '19 at 1:39