5
\$\begingroup\$

The goal here is to pull all alerts from a single iterable:

obj = Alerts(db, args)
for alert in obj.alerts()
    pass

Now, I need to add a few more sources and I'm not sure if this is a good approach to instantiate all classes in the AllAlerts constructor? I also don't like the fact that I'll have to add them to the self.sources attribute each time there is a new one (achieving loose coupling).

Based on the code snippet provided below could you recommend some different, perhaps a better approach?

Code:

from itertools import chain

from . import mappings
from .utils import converter


class BaseSource(object):
    def __init__(self, db, args):
        self.args = args
        self.db = db

    def alerts(self):
        raise NotImplementedError

    def _data(self, mapping, source):
        """
        This method do the parsing based on data source.
        """
        for entry in self._raw_data():
            yield converter(source, mapping, entry)


class NagiosSource(BaseSource):
    def __init__(self, *args, **kwargs):
        ...
        super().__init__(*args, **kwargs)

    def _raw_data(self):
        """
        The logic to get the data from Nagios.
        """

    def alerts(self):
        mapping = mappings.nagios
        return self._data(mapping, "nagios")


class ZabbixSource(BaseSource):
    def __init__(self, *args, **kwargs):
        ...
        super().__init__(*args, **kwargs)

    def _raw_data(self):
        """
        The logic to get the data from Zabbix.
        """

    def alerts(self):
        mapping = mappings.zabbix
        return self._data(mapping, "zabbix")


class AllAlerts(BaseSource):
    def __init__(self, db, args):
        self.sources = (
            NagiosSource(db, args),
            ZabbixSource(db, args),
        )
        super().__init__(db, args)

    def alerts(self):
        return chain.from_iterable(s.data() for s in self.sources)

I though about adding a class decorator that will register all sources but then again, I would have to use a global variable and not sure how I could pass args when creating objects...

test.py:

sources = set()

def register(cls):
    sources.add(cls())
    return cls


@register
class NagiosSource:
    pass


@register
class ZabbixSource:
    pass

Test:

$ python test.py
{<__main__.ZabbixSource object at 0x7f1a3b1d26d0>, <__main__.NagiosSource object at 0x7f1a3b1d2760>}
\$\endgroup\$
3
\$\begingroup\$

Recording Subclasses

As of Python 3.6, there is an easy way to gather all your subclasses together, without having to risk the error-prone method of manually creating an AllAlerts subclass and listing all of the subclasses in it. The key is object.__init_subclass__(cls). It is called when a subclass is defined.

class BaseSource:
    subclasses = []

    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        cls.subclasses.append(cls)
        print("Registered", cls)

Now, whenever a subclass of BaseSource is defined, that subclass will be added to the BaseSource.subclass list.

Of course, AllAlerts did more than this. It created one instance of each source subclass and passed the same arguments in the constructor of each. We'll have to do that in a @classmethod of the base class. It also used itertools to chain together all of the alerts from each of those source instances, so we'll have to record those source instances, and provide a @classmethod for getting that chain of alerts.

from itertools import chain

class BaseSource:
    subclasses = []

    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        cls.subclasses.append(cls)
        print("Registered", cls)

    @classmethod
    def init_all(cls, db, args):
        """
        Create one instance for each subclass, constructed using the given
        'db' and 'args' values.
        """
        cls.sources = (subclass(db, args) for subclass in cls.subclasses)

    @classmethod
    def all_alerts(cls):
        """
        Return an iterable of all alerts from all subclass sources
        """
        return chain.from_iterable(src.alerts() for src in cls.sources)

    def __init__(self, db, args):
        self.db = db
        self.args = args

    def alerts(self):
        """
        Return an iterable of alerts for this class
        """
        raise NotImplementedError()

With this base class, you just need to define the source subclasses, as many as you like. There is no need to remember all of the classes; the base class does that for you:

class NagiosSource(BaseSource):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        print("Constructed Nagios Source")

    def alerts(self):
        yield "Alert 1"
        yield "Alert 2"


class ZabbixSource(BaseSource):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        print("Constructed Zabbix Source")

    def alerts(self):
        yield "Alert A"
        yield "Alert B"

After all of the subclass definitions have been read in, you will have to initialize them with the appropriate db and args, like you created an AllAlerts instance, which created all of the source objects. And then you can request all alerts from the base class:

BaseSource.init_all("mydb", (1, 2, 3))

print(list(BaseSource.all_alerts()))

Output of above:

Registered <class '__main__.NagiosSource'>
Registered <class '__main__.ZabbixSource'>
Constructed Nagios Source
Constructed Zabbix Source
['Alert 1', 'Alert 2', 'Alert A', 'Alert B']
>>> 

Raise objects, not classes

Your BaseSource had the method:

    def alerts(self):
        raise NotImplementedError

This appears to be raise a class instead of an instance of a class. You should write:

    def alerts(self):
        raise NotImplementedError()

Using instances allows you to have arguments, which helps describe the error. What does "Not Implemented" mean? Does it mean "Not Implemented Yet", as in a later version of the library is expected to provide an implementation? No! We need subclasses to provide the implementation.

    def alerts(self):
        raise NotImplementedError("This method must be overridden in derived classes")

Public methods need docstrings

You provide docstrings for _data() and _raw_data(), but not for alerts(). This is backwards.

The leading underscore represents private methods. An external caller does not need to know how to call them, because they are private.

On the other hand, public functions (without the leading underscore) are expected to be called by external callers. And someone writing the code which uses these Source objects may want to know how to call the methods. So they may type:

>>> help(NagiosSource)

and would reasonable expect to get information about how to use the class and its public method. The docstring for _data and _raw_data would not be provided, because of the leading underscore.

You may provide docstrings for private methods (the expectation is the public methods would all have been documented first), but code comments may be just as useful.

\$\endgroup\$

Your Answer

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

Not the answer you're looking for? Browse other questions tagged or ask your own question.