add function for loading plugin modules

Loads modules for configurable code like tokenizers, sanitizers, etc.
Supports internal modules, external libraries and code from the
project directory.
This commit is contained in:
Sarah Hoffmann
2022-07-25 15:17:20 +02:00
parent 95d4061b2a
commit 7b7203c149
2 changed files with 122 additions and 0 deletions

View File

@@ -8,8 +8,10 @@
Nominatim configuration accessor.
"""
from typing import Dict, Any, List, Mapping, Optional
import importlib.util
import logging
import os
import sys
from pathlib import Path
import json
import yaml
@@ -73,6 +75,7 @@ class Configuration:
data: Path
self.lib_dir = _LibDirs()
self._private_plugins: Dict[str, object] = {}
def set_libdirs(self, **kwargs: StrPath) -> None:
@@ -219,6 +222,49 @@ class Configuration:
return result
def load_plugin_module(self, module_name: str, internal_path: str) -> object:
""" Load a Python module as a plugin.
The module_name may have three variants:
* A name without any '.' is assumed to be an internal module
and will be searched relative to `internal_path`.
* If the name ends in `.py`, module_name is assumed to be a
file name relative to the project directory.
* Any other name is assumed to be an absolute module name.
In either of the variants the module name must start with a letter.
"""
if not module_name or not module_name[0].isidentifier():
raise UsageError(f'Invalid module name {module_name}')
if '.' not in module_name:
module_name = module_name.replace('-', '_')
full_module = f'{internal_path}.{module_name}'
return sys.modules.get(full_module) or importlib.import_module(full_module)
if module_name.endswith('.py'):
if self.project_dir is None or not (self.project_dir / module_name).exists():
raise UsageError(f"Cannot find module '{module_name}' in project directory.")
if module_name in self._private_plugins:
return self._private_plugins[module_name]
file_path = str(self.project_dir / module_name)
spec = importlib.util.spec_from_file_location(module_name, file_path)
if spec:
module = importlib.util.module_from_spec(spec)
# Do not add to global modules because there is no standard
# module name that Python can resolve.
self._private_plugins[module_name] = module
assert spec.loader is not None
spec.loader.exec_module(module)
return module
return sys.modules.get(module_name) or importlib.import_module(module_name)
def find_config_file(self, filename: StrPath,
config: Optional[str] = None) -> Path:
""" Resolve the location of a configuration file given a filename and

View File

@@ -0,0 +1,76 @@
# SPDX-License-Identifier: GPL-2.0-only
#
# This file is part of Nominatim. (https://nominatim.org)
#
# Copyright (C) 2022 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Test for loading extra Python modules.
"""
from pathlib import Path
import sys
import pytest
from nominatim.config import Configuration
@pytest.fixture
def test_config(src_dir, tmp_path):
""" Create a configuration object with project and config directories
in a temporary directory.
"""
(tmp_path / 'project').mkdir()
(tmp_path / 'config').mkdir()
conf = Configuration(tmp_path / 'project', src_dir / 'settings')
conf.config_dir = tmp_path / 'config'
return conf
def test_load_default_module(test_config):
module = test_config.load_plugin_module('version', 'nominatim')
assert isinstance(module.NOMINATIM_VERSION, tuple)
def test_load_plugin_module(test_config, tmp_path):
(tmp_path / 'project' / 'testpath').mkdir()
(tmp_path / 'project' / 'testpath' / 'mymod.py')\
.write_text("def my_test_function():\n return 'gjwitlsSG42TG%'")
module = test_config.load_plugin_module('testpath/mymod.py', 'private.something')
assert module.my_test_function() == 'gjwitlsSG42TG%'
# also test reloading module
(tmp_path / 'project' / 'testpath' / 'mymod.py')\
.write_text("def my_test_function():\n return 'hjothjorhj'")
module = test_config.load_plugin_module('testpath/mymod.py', 'private.something')
assert module.my_test_function() == 'gjwitlsSG42TG%'
def test_load_external_library_module(test_config, tmp_path, monkeypatch):
MODULE_NAME = 'foogurenqodr4'
pythonpath = tmp_path / 'priv-python'
pythonpath.mkdir()
(pythonpath / MODULE_NAME).mkdir()
(pythonpath / MODULE_NAME / '__init__.py').write_text('')
(pythonpath / MODULE_NAME / 'tester.py')\
.write_text("def my_test_function():\n return 'gjwitlsSG42TG%'")
monkeypatch.syspath_prepend(pythonpath)
module = test_config.load_plugin_module(f'{MODULE_NAME}.tester', 'private.something')
assert module.my_test_function() == 'gjwitlsSG42TG%'
# also test reloading module
(pythonpath / MODULE_NAME / 'tester.py')\
.write_text("def my_test_function():\n return 'dfigjreigj'")
module = test_config.load_plugin_module(f'{MODULE_NAME}.tester', 'private.something')
assert module.my_test_function() == 'gjwitlsSG42TG%'
del sys.modules[f'{MODULE_NAME}.tester']