Source code for ucsschool.importer.utils.import_pyhook

#!/usr/bin/python3
# -*- coding: utf-8 -*-
#
# Univention UCS@school
# Copyright 2017-2021 Univention GmbH
#
# https://www.univention.de/
#
# All rights reserved.
#
# The source code of this program is made available
# under the terms of the GNU Affero General Public License version 3
# (GNU AGPL V3) as published by the Free Software Foundation.
#
# Binary versions of this program provided by Univention to you as
# well as other copyrighted, protected or trademarked materials like
# Logos, graphics, fonts, specific documentations and configurations,
# cryptographic keys etc. are subject to a license agreement between
# you and Univention and not subject to the GNU AGPL V3.
#
# In the case you use this program under the terms of the GNU AGPL V3,
# the program is provided in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public
# License with the Debian GNU/Linux or Univention distribution in file
# /usr/share/common-licenses/AGPL-3; if not, see
# <http://www.gnu.org/licenses/>.

"""
Base class for all Python based import hooks.
"""

from __future__ import absolute_import

import logging
from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Type, TypeVar, Union

from ucsschool.lib.pyhooks import PyHook, PyHooksLoader

from ..exceptions import InitialisationError
from .ldap_connection import get_admin_connection, get_readonly_connection

if TYPE_CHECKING:
    import univention.admin.uldap

ImportPyHookTV = TypeVar("ImportPyHookTV", bound="ImportPyHook")


__import_pyhook_loader_instance = None  # type: Optional["ImportPyHookLoader"]


[docs]class ImportPyHook(PyHook): """ Base class for Python based import hooks. * self.dry_run # whether hook is executed during a dry-run (1) * self.lo # LDAP connection object (2) * self.logger # Python logging instance If multiple hook classes are found, hook functions with higher priority numbers run before those with lower priorities. None disables a function (no need to remove it / comment it out). (1) Hooks are only executed during dry-runs, if the class attribute :py:attr:`supports_dry_run` is set to `True` (default is `False`). Hooks with `supports_dry_run == True` must not modify LDAP objects. Therefore the LDAP connection object self.lo will be a read-only connection during a dry-run. (2) Read-write cn=admin connection in a real run, read-only cn=admin connection during a dry-run. """ supports_dry_run = False # if True hook will be executed during a dry-run def __init__(self, lo=None, dry_run=None, *args, **kwargs): # type: (Optional[univention.admin.uldap.access], Optional[bool], *Any, **Any) -> None """ :param univention.admin.uldap.access lo: optional LDAP connection object :param bool dry_run: whether hook is executed during a dry-run """ super(ImportPyHook, self).__init__(*args, **kwargs) if dry_run is None: from ..configuration import Configuration try: config = Configuration() self.dry_run = config["dry_run"] """Whether this is a dry-run""" except InitialisationError: self.dry_run = False else: self.dry_run = dry_run if lo is None: self.lo = ( get_readonly_connection()[0] if self.dry_run else get_admin_connection()[0] ) # type: univention.admin.uldap.access """LDAP connection object""" else: self.lo = lo # reuse LDAP object """LDAP connection object""" self.logger = logging.getLogger(__name__) # type: logging.Logger """Python logging instance"""
[docs]class ImportPyHookLoader(object): """ Load and initialize hooks. If hooks should be instantiated with arguments, use :py:meth:`init_hook()` before :py:meth:`call_hook()`. """ _pyhook_obj_cache = {} # type: Dict[Type[ImportPyHookTV], Dict[str, List[Callable[..., Any]]]] def __init__(self, pyhooks_base_path): # type: (str) -> None self.pyhooks_base_path = pyhooks_base_path self.logger = logging.getLogger(__name__) # type: logging.Logger
[docs] def init_hook(self, hook_cls, filter_func=None, *args, **kwargs): # type: (Union[Type[ImportPyHookTV], str], Optional[Callable[[Type[ImportPyHookTV]], bool]], *Any, **Any) -> Dict[str, List[Callable[..., Any]]] # noqa: E501 """ Load and initialize hook class `hook_cls`. :param hook_cls: class object - load and run hooks that are a subclass of this :type hook_cls: ucsschool.importer.utils.import_pyhook.ImportPyHook or str :param filter_func: function to filter out undesired hook classes (takes a class and returns a bool), passed to PyHooksLoader :type filter_func: callable or None :param tuple args: arguments to pass to __init__ of hooks :param dict kwargs: arguments to pass to __init__ of hooks :return: mapping from method names to list of methods of initialized hook objects, sorted by method priority :rtype: dict[str, list[callable]] """ # The PyHook objects themselves are already cached by PyHooksLoader, but we don't want to # initialize a PyHooksLoader each time we run a hook, so we'll keep a dict linking directly to # all PyHooksLoader caches. base_class = PyHooksLoader.hook_cls2importpyhook(hook_cls, "hook_cls") if not issubclass(base_class, ImportPyHook): raise TypeError("Argument 'hook_cls' must be a subclass of ImportPyHook.") if base_class not in self._pyhook_obj_cache: pyhooks_loader = PyHooksLoader(self.pyhooks_base_path, base_class, self.logger, filter_func) self._pyhook_obj_cache[base_class] = pyhooks_loader.get_hook_objects(*args, **kwargs) return self._pyhook_obj_cache[base_class]
[docs] def call_hooks(self, hook_cls, func_name, *args, **kwargs): # type: (Type[ImportPyHookTV], str, *Any, **Any) -> List[Any] """ Run hooks with name `func_name` from class `hook_cls`. :param hook_cls: class object - load and run hooks that are a subclass of this :param str func_name: name of method to run in each hook :param args: arguments to pass to hook function :param kwargs: arguments to pass to hook function :return: list of when all executed hooks returned :rtype: list """ hooks = self.init_hook(hook_cls) res = [] for func in hooks.get(func_name, []): self.logger.info("Running %s %s hook %s ...", self.__class__.__name__, func_name, func) res.append(func(*args, **kwargs)) return res
[docs]def get_import_pyhooks(hook_cls, filter_func=None, *args, **kwargs): # type: (Union[Type[ImportPyHookTV], str], Optional[Callable[[Type[ImportPyHookTV]], bool]], *Any, **Any) -> Dict[str, List[Callable[..., Any]]] # noqa: E501 """ Retrieve (and initialize subclasses of :py:class:`hook_cls`, if not yet done) pyhooks of type `hook_cls`. Results are cached. If no argument must be passed to the `hook_cls` :py:meth:`__init__()` or :py:class:`PyHooksLoader`, then it is not necessary to call this function, just use :py:func:`run_import_pyhooks` directly. Convenience function for easy usage of PyHooksLoader. :param hook_cls: class object or fully dotted Python path to a class definition - load and run hooks that are a subclass of this :type hook_cls: ucsschool.importer.utils.import_pyhook.ImportPyHook or str :param filter_func: function to filter out undesired hook classes (takes a class and returns a bool), passed to PyHooksLoader :type filter_func: callable or None :param args: arguments to pass to __init__ of hooks :param kwargs: arguments to pass to __init__ of hooks :return: mapping from method names to list of methods of initialized hook objects, sorted by method priority :rtype: dict[str, list[callable]] """ global __import_pyhook_loader_instance if not __import_pyhook_loader_instance: from ..configuration import Configuration try: config = Configuration() path = config.get("hooks_dir_pyhook", "/usr/share/ucs-school-import/pyhooks") if "dry_run" not in kwargs: kwargs["dry_run"] = config["dry_run"] except InitialisationError: path = "/usr/share/ucs-school-import/pyhooks" __import_pyhook_loader_instance = ImportPyHookLoader(path) return __import_pyhook_loader_instance.init_hook(hook_cls, filter_func, *args, **kwargs)
[docs]def run_import_pyhooks(hook_cls, func_name, *args, **kwargs): # type: (Union[Type[ImportPyHookTV], str], str, *Any, **Any) -> List[Any] """ Execute method `func_name` of subclasses of `hook_cls`, load and initialize if required. Convenience function for easy usage of PyHooksLoader. :param hook_cls: class object or fully dotted Python path to a class definition - load and run hooks that are a subclass of this :type hook_cls: ucsschool.importer.utils.import_pyhook.ImportPyHook or str :param str func_name: name of method to run in each hook :param args: arguments to pass to hook function `func_name` :param kwargs: arguments to pass to hook function `func_name` :return: list of when all executed hooks returned :rtype: list """ get_import_pyhooks(hook_cls) return __import_pyhook_loader_instance.call_hooks(hook_cls, func_name, *args, **kwargs)