Source code for anitya.lib.plugins

# -*- coding: utf-8 -*-
# This file is a part of the Anitya project.
# Copyright © 2014-2017 Pierre-Yves Chibon <>
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# GNU General Public License for more details.
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
"""Module handling the load/call of the plugins of anitya."""

import logging

from straight.plugin import load

from anitya.lib.backends import BaseBackend
from anitya.lib.ecosystems import BaseEcosystem
from anitya.lib.versions import Version

_log = logging.getLogger(__name__)

class _PluginManager(object):
    """Manage a particular set of Anitya plugins"""

    def __init__(self, namespace, base_class):
        self._namespace = namespace
        self._base_class = base_class

    def get_plugins(self):
        """Return the list of plugins."""
        return load(self._namespace, subclasses=self._base_class)

    def get_plugin_names(self):
        """Return the list of plugin names."""
        plugins = self.get_plugins()
        output = [ for plugin in plugins]
        return output

    def get_plugin(self, plugin_name):
        """Return the plugin corresponding to the given plugin name."""
        plugins = self.get_plugins()
        for plugin in plugins:
            if == plugin_name.lower():
                return plugin

BACKEND_PLUGINS = _PluginManager("anitya.lib.backends", BaseBackend)
ECOSYSTEM_PLUGINS = _PluginManager("anitya.lib.ecosystems", BaseEcosystem)
VERSION_PLUGINS = _PluginManager("anitya.lib.versions", Version)

def _load_backend_plugins(session):
    """Load any new backend plugins into the DB"""
    plugins = list(BACKEND_PLUGINS.get_plugins())
    return plugins

def _load_ecosystem_plugins(session):
    """Load any new ecosystem plugins into the DB"""
    plugins = list(ECOSYSTEM_PLUGINS.get_plugins())
    return plugins

def _load_version_plugins(session):
    """Load any new version plugins into the DB"""
    plugins = list(VERSION_PLUGINS.get_plugins())
    return plugins

[docs]def load_all_plugins(session): """Load all the plugins and insert them in the database if they are not already present.""" plugins = {} plugins["backends"] = _load_backend_plugins(session) plugins["ecosystems"] = _load_ecosystem_plugins(session) plugins["versions"] = _load_version_plugins(session) return plugins
# Preserve module level API for accessing the backend plugin list get_plugin_names = BACKEND_PLUGINS.get_plugin_names get_plugins = BACKEND_PLUGINS.get_plugins get_plugin = BACKEND_PLUGINS.get_plugin
[docs]def load_plugins(session, family="backends"): """Calls load_all_plugins, but only retuns plugins specified by family argument Args: family (str): family of the plugins, that should be returned """ return load_all_plugins(session)[family]