You have several questions, so let us take a look at them in the order they were asked:
- The closest thing to a
Bag that you can find in Python's standard library is collections.Counter. However, it is not designed for concurrent usage and has different methods than the ConcurrentBag class you are familiar with. By inheriting from collection.Counter and specifying a custom metaclass designed to make classes thread-safe, it is possible to easily recreate the ConcurrentBag class for use in Python.
- The getter for your thread-safe collection can be duplicated in Python using the
property decorator. Since you are making the information read-only, this is the easiest method to implement. You can see the property in the code below just below the initializer for the Message class.
- Yes, it is possible to check if two objects are equal to each other in Python. You have to define a special
__eq__ method as part of the class. The version shown in the example code is just below the aforementioned property. It first checks the type of the instance being compared, and then looks through all the other attributes of each instance except for __time.
- Python has no native concept of sealing a class, but a simple metaclass can provide very similar functionality. By setting the metaclass of a class to
SealedMeta, any classes that try to inherit from from the "sealed" class will cause a RuntimeError to be raised. Please be aware, though, that anyone with access to the source code of your class could remove its sealed designation.
The following classes and metaclasses should solve the issues you were facing. They are as follows:
SealedMeta is a metaclass that can be used to mark a class as being sealed. Any class that tries to inherit from a sealed class will fail to be constructed since a RuntimeError exception will be generated. Because the requirements are so simple, this metaclass is significantly simpler than the second shown in the example code.Message is an attempt at recreating the class you wrote in C#. The arguments and attributes have slightly different names, but the same basic idea is present. Since types are not statically specified in Python, only one equality-checking method is needed in this class. Note how the class is marked as sealed with (metaclass=SealedMeta) at the top.AtomicMeta was a challenging metaclass to write, and I am not sure that it is completely correct. Not only does it have to make the methods of a class thread-safe, it must process all bases that the class inherits from. Since old-style super calls may be present in those bases, the metaclass attempts to fix the modules they are located in.ConcurrentBag inherits from collections.Counter since it is most like a bag from other languages. Since you wanted the class to be thread-safe, it must use the AtomicMeta metaclass so that the methods of it and its parents are changed into atomic operations. A to_array method is introduced to improve its API.
Without further delay, here is the code referenced above. Hopefully, it will be a help both to you and others:
#! /usr/bin/env python3
import builtins
import collections
import functools
import inspect
import threading
class SealedMeta(type):
"""SealedMeta(name, bases, dictionary) -> new sealed class"""
__REGISTRY = ()
def __new__(mcs, name, bases, dictionary):
"""Create a new class only if it is not related to a sealed class."""
if any(issubclass(base, mcs.__REGISTRY) for base in bases):
raise RuntimeError('no class may inherit from a sealed class')
mcs.__REGISTRY += (super().__new__(mcs, name, bases, dictionary),)
return mcs.__REGISTRY[-1]
class Message(metaclass=SealedMeta):
"""Message(identifier, message, source, kind, time) -> Message instance"""
def __init__(self, identifier, message, source, kind, time):
"""Initialize all the attributes in a new Message instance."""
self.__identifier = identifier
self.__message = message
self.__source = source
self.__kind = kind
self.__time = time
self.__destination = ConcurrentBag()
@property
def destination(self):
"""Destination property containing serialized Employee instances."""
return self.__destination.to_array()
def __eq__(self, other):
"""Return if this instance has the same data as the other instance."""
# noinspection PyPep8
return isinstance(other, type(self)) and \
self.__identifier == other.__identifier and \
self.__message == other.__message and \
self.__source == other.__source and \
self.__kind == other.__kind and \
self.__destination == other.__destination
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
))
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
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:
instance_lock = self.__lock = threading.RLock()
with instance_lock:
return method(self, *args, **kwargs)
return atomic_wrapper
# noinspection PyAbstractClass
class ConcurrentBag(collections.Counter, metaclass=AtomicMeta):
"""ConcurrentBag() -> ConcurrentBag instance"""
def to_array(self):
"""Serialize the data in the ConcurrentBag instance."""
return tuple(key for key, value in self.items() for _ in range(value))
After doing a little research, you might find that there are alternatives for sealing a class. In Java, classes are marked final instead of using a sealed keyword. Searching for the alternative keyword could bring you to Martijn Pieters' answer which introduces a slightly simpler metaclass for sealing classes in Python. You can use the following metaclass inspired by his the same way that SealedMeta is currently used in the code:
class Final(type):
"""Final(name, bases, dictionary) -> new final class"""
def __new__(mcs, name, bases, dictionary):
"""Create a new class if none of its bases are marked as final."""
if any(isinstance(base, mcs) for base in bases):
raise TypeError('no class may inherit from a final class')
return super().__new__(mcs, name, bases, dictionary)
The Final metaclass is useful if you want to seal a class and prevent others from inheriting from it, but a different approach is needed if attributes of a class are intended to be final instead. In such a case, the Access metaclass can be rather useful. It was inspired by this answer to the question Prevent function overriding in Python. A modified approach is taken in the metaclass shown below that is designed to take into account attributes of all kinds.
import collections
import threading
# noinspection PyProtectedMember
class RLock(threading._RLock):
"""RLock() -> RLock instance with count property"""
@property
def count(self):
"""Count property showing current level of lock ownership."""
return self._count
class Access(type):
"""Access(name, bases, dictionary) -> class supporting final attributes"""
__MUTEX = RLock()
__FINAL = []
@classmethod
def __prepare__(mcs, name, bases, **keywords):
"""Begin construction of a class and check for possible deadlocks."""
if not mcs.__MUTEX.acquire(True, 10):
raise RuntimeError('please check your code for deadlocks')
# noinspection PyUnresolvedReferences
return super().__prepare__(mcs, name, bases, **keywords)
@classmethod
def final(mcs, attribute):
"""Record an attribute as being final so it cannot be overridden."""
with mcs.__MUTEX:
if any(attribute is final for final in mcs.__FINAL):
raise SyntaxError('attributes may be marked final only once')
mcs.__FINAL.append(attribute)
return attribute
def __new__(mcs, class_name, bases, dictionary):
"""Create a new class that supports the concept of final attributes."""
classes, visited, names = collections.deque(bases), set(), set()
# Find all attributes marked as final in base classes.
while classes:
base = classes.popleft()
if base not in visited:
visited.add(base)
classes.extend(base.__bases__)
names.update(getattr(base, '__final_attributes__', ()))
# Verify that the current class does not override final attributes.
if any(name in names for name in dictionary):
raise SyntaxError('final attributes may not be overridden')
names.clear()
# Collect the names of all attributes that are marked as final.
for name, attribute in dictionary.items():
for index, final in enumerate(mcs.__FINAL):
if attribute is final:
del mcs.__FINAL[index]
names.add(name)
break
# Do a sanity check to ensure this metaclass is being used properly.
if mcs.__MUTEX.count == 1 and mcs.__FINAL:
raise RuntimeError('final decorator has not been used correctly')
mcs.__MUTEX.release()
dictionary['__final_attributes__'] = frozenset(names)
return super().__new__(mcs, class_name, bases, dictionary)
As a demonstration of how to use the Access metaclass, this example will raise a SyntaxError:
class Parent(metaclass=Access):
def __init__(self, a, b):
self.__a = a
self.__b = b
@Access.final
def add(self):
return self.__a + self.__b
def sub(self):
return self.__a - self.__b
class Child(Parent):
def __init__(self, a, b, c):
super().__init__(a, b)
self.__c = c
def add(self):
return super().add() + self.__c
