Source code for Bcfg2.Server.Plugins.Packages.PackagesSources

""" PackagesSources handles the
:ref:`server-plugins-generators-packages` ``sources.xml`` file"""

import os
import sys
import Bcfg2.Server.Plugin
from Bcfg2.Server.Statistics import track_statistics
from Bcfg2.Server.Plugins.Packages.Source import SourceInitError


# pylint: disable=E0012,R0924
[docs]class PackagesSources(Bcfg2.Server.Plugin.StructFile): """ PackagesSources handles parsing of the :mod:`Bcfg2.Server.Plugins.Packages` ``sources.xml`` file, and the creation of the appropriate :class:`Bcfg2.Server.Plugins.Packages.Source.Source` object for each ``Source`` tag. """ __identifier__ = None create = "Sources" def __init__(self, filename, cachepath, packages): """ :param filename: The full path to ``sources.xml`` :type filename: string :param cachepath: The full path to the directory where :class:`Bcfg2.Server.Plugins.Packages.Source.Source` data will be cached :type cachepath: string :param packages: The Packages plugin object ``sources.xml`` is being parsed on behalf of (i.e., the calling object) :type packages: Bcfg2.Server.Plugins.Packages.Packages :raises: :class:`Bcfg2.Server.Plugin.exceptions.PluginInitError` - If ``sources.xml`` cannot be read """ Bcfg2.Server.Plugin.StructFile.__init__(self, filename, should_monitor=True) #: The full path to the directory where #: :class:`Bcfg2.Server.Plugins.Packages.Source.Source` data #: will be cached self.cachepath = cachepath if not os.path.exists(self.cachepath): # create cache directory if needed try: os.makedirs(self.cachepath) except OSError: err = sys.exc_info()[1] self.logger.error("Could not create Packages cache at %s: %s" % (self.cachepath, err)) #: The :class:`Bcfg2.Server.Plugins.Packages.Packages` that #: instantiated this ``PackagesSources`` object self.pkg_obj = packages #: The set of all XML files that have been successfully #: parsed. This is used by :attr:`loaded` to determine if the #: sources have been fully parsed and the #: :class:`Bcfg2.Server.Plugins.Packages.Packages` plugin #: should be told to reload its data. self.parsed = set()
[docs] def set_debug(self, debug): Bcfg2.Server.Plugin.StructFile.set_debug(self, debug) for source in self.entries: source.set_debug(debug)
[docs] def HandleEvent(self, event=None): """ HandleEvent is called whenever the FAM registers an event. When :attr:`loaded` becomes True, :func:`Bcfg2.Server.Plugins.Packages.Packages.Reload` is called to reload all plugin data from the configured sources. :param event: The event object :type event: Bcfg2.Server.FileMonitor.Event :returns: None """ if event and event.filename != self.name: for fpath in self.extras: if fpath == os.path.abspath(event.filename): self.parsed.add(fpath) break Bcfg2.Server.Plugin.StructFile.HandleEvent(self, event=event) if self.loaded: self.logger.info("Reloading Packages plugin") self.pkg_obj.Reload()
@property def loaded(self): """ Whether or not all XML files (``sources.xml`` and everything XIncluded in it) have been parsed. This flag is used to determine if the Packages plugin should be told to load its data. """ return sorted(list(self.parsed)) == sorted(self.extras)
[docs] @track_statistics() def Index(self): Bcfg2.Server.Plugin.StructFile.Index(self) self.entries = [] if self.loaded: for xsource in self.xdata.findall('.//Source'): source = self.source_from_xml(xsource) if source is not None: self.entries.append(source)
Index.__doc__ = Bcfg2.Server.Plugin.StructFile.Index.__doc__ + """ ``Index`` is responsible for calling :func:`source_from_xml` for each ``Source`` tag in each file. """
[docs] @track_statistics() def source_from_xml(self, xsource): """ Create a :class:`Bcfg2.Server.Plugins.Packages.Source.Source` subclass object from XML representation of a source in ``sources.xml``. ``source_from_xml`` determines the appropriate subclass of ``Source`` to instantiate according to the ``type`` attribute of the ``Source`` tag. :param xsource: The XML tag representing the source :type xsource: lxml.etree._Element :returns: :class:`Bcfg2.Server.Plugins.Packages.Source.Source` subclass, or None on error """ stype = xsource.get("type") if stype is None: self.logger.error("Packages: No type specified for source at %s, " "skipping" % (xsource.get("rawurl", xsource.get("url")))) return None cls = None for mod in Bcfg2.Options.setup.packages_backends: if mod.__name__.endswith(".%s" % stype.title()): cls = getattr(mod, "%sSource" % stype.title()) break else: self.logger.error("Packages: Unknown source type %s" % stype) return None try: source = cls(self.cachepath, xsource) except SourceInitError: err = sys.exc_info()[1] self.logger.error("Packages: %s" % err) source = None return source
def __getitem__(self, key): return self.entries[key] def __repr__(self): return "PackagesSources: %s" % repr(self.entries) def __str__(self): return "PackagesSources: %s sources" % len(self.entries) def __len__(self): return len(self.entries)