Skip to content

Plugins API

Plugin registry and discovery for custom validators and exporters.

PluginRegistry

Central registry for validator and exporter plugins.

dppvalidator.plugins.PluginRegistry

Registry for validator and exporter plugins.

Automatically discovers plugins via entry points on initialization, or allows manual registration for testing and custom setups.

Source code in src/dppvalidator/plugins/registry.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
class PluginRegistry:
    """Registry for validator and exporter plugins.

    Automatically discovers plugins via entry points on initialization,
    or allows manual registration for testing and custom setups.
    """

    def __init__(self, auto_discover: bool = True) -> None:
        """Initialize plugin registry.

        Args:
            auto_discover: If True, discover plugins via entry points
        """
        self._validators: dict[str, type[SemanticRule] | SemanticRule] = {}
        self._exporters: dict[str, type[Exporter] | Exporter] = {}

        if auto_discover:
            self._discover_all()

    def _discover_all(self) -> None:
        """Discover all plugins via entry points."""
        for name, validator in discover_validators():
            self._validators[name] = validator
            logger.info("Registered validator plugin: %s", name)

        for name, exporter in discover_exporters():
            self._exporters[name] = exporter
            logger.info("Registered exporter plugin: %s", name)

    def register_validator(self, name: str, validator: type[SemanticRule] | SemanticRule) -> None:
        """Register a validator plugin.

        Args:
            name: Unique name for the validator
            validator: Validator class or instance implementing SemanticRule protocol
        """
        self._validators[name] = validator
        logger.debug("Manually registered validator: %s", name)

    def register_exporter(self, name: str, exporter: type[Exporter] | Exporter) -> None:
        """Register an exporter plugin.

        Args:
            name: Unique name for the exporter
            exporter: Exporter class or instance implementing Exporter protocol
        """
        self._exporters[name] = exporter
        logger.debug("Manually registered exporter: %s", name)

    def unregister_validator(self, name: str) -> bool:
        """Unregister a validator plugin.

        Args:
            name: Name of validator to unregister

        Returns:
            True if validator was found and removed
        """
        return self._validators.pop(name, None) is not None

    def unregister_exporter(self, name: str) -> bool:
        """Unregister an exporter plugin.

        Args:
            name: Name of exporter to unregister

        Returns:
            True if exporter was found and removed
        """
        return self._exporters.pop(name, None) is not None

    def get_validator(self, name: str) -> type[SemanticRule] | SemanticRule | None:
        """Get a validator by name.

        Args:
            name: Validator name

        Returns:
            Validator or None if not found
        """
        return self._validators.get(name)

    def get_exporter(self, name: str) -> type[Exporter] | Exporter | None:
        """Get an exporter by name.

        Args:
            name: Exporter name

        Returns:
            Exporter or None if not found
        """
        return self._exporters.get(name)

    def run_all_validators(
        self,
        passport: DigitalProductPassport,
        *,
        strict: bool = False,
    ) -> list[ValidationError]:
        """Run all registered validator plugins.

        Args:
            passport: Parsed passport to validate
            strict: If True, raise PluginError on plugin failures instead of
                returning a warning. Useful for CI/CD pipelines.

        Returns:
            List of validation errors from all plugins

        Raises:
            PluginError: If strict=True and a plugin fails to execute
        """
        errors: list[ValidationError] = []

        for name, validator in self._validators.items():
            try:
                instance = validator() if isinstance(validator, type) else validator

                if hasattr(instance, "check"):
                    violations = instance.check(passport)
                    rule_id = getattr(instance, "rule_id", f"PLG_{name.upper()}")
                    severity = getattr(instance, "severity", "error")

                    for path, message in violations:
                        errors.append(
                            ValidationError(
                                path=path,
                                message=message,
                                code=rule_id,
                                layer="plugin",
                                severity=severity,
                            )
                        )

            except (AttributeError, TypeError, ValueError, RuntimeError) as e:
                if strict:
                    raise PluginError(f"Plugin '{name}' failed: {e}") from e
                logger.warning("Plugin %s failed: %s", name, e)
                errors.append(
                    ValidationError(
                        path="$",
                        message=f"Plugin '{name}' failed: {e}",
                        code="PLG001",
                        layer="plugin",
                        severity="warning",
                    )
                )

        return errors

    @property
    def validator_names(self) -> list[str]:
        """List of registered validator names."""
        return list(self._validators.keys())

    @property
    def exporter_names(self) -> list[str]:
        """List of registered exporter names."""
        return list(self._exporters.keys())

    @property
    def validator_count(self) -> int:
        """Number of registered validators."""
        return len(self._validators)

    @property
    def exporter_count(self) -> int:
        """Number of registered exporters."""
        return len(self._exporters)

exporter_count property

Number of registered exporters.

exporter_names property

List of registered exporter names.

validator_count property

Number of registered validators.

validator_names property

List of registered validator names.

__init__(auto_discover=True)

Initialize plugin registry.

Parameters:

Name Type Description Default
auto_discover bool

If True, discover plugins via entry points

 True
Source code in src/dppvalidator/plugins/registry.py 
31
32
33
34
35
36
37
38
39
40
41
def __init__(self, auto_discover: bool = True) -> None:
    """Initialize plugin registry.

    Args:
        auto_discover: If True, discover plugins via entry points
    """
    self._validators: dict[str, type[SemanticRule] | SemanticRule] = {}
    self._exporters: dict[str, type[Exporter] | Exporter] = {}

    if auto_discover:
        self._discover_all()

get_exporter(name)

Get an exporter by name.

Parameters:

Name Type Description Default
name str

Exporter name

 required

Returns:

Type Description
type[Exporter] | Exporter | None

Exporter or None if not found
Source code in src/dppvalidator/plugins/registry.py 
106
107
108
109
110
111
112
113
114
115
def get_exporter(self, name: str) -> type[Exporter] | Exporter | None:
    """Get an exporter by name.

    Args:
        name: Exporter name

    Returns:
        Exporter or None if not found
    """
    return self._exporters.get(name)

get_validator(name)

Get a validator by name.

Parameters:

Name Type Description Default
name str

Validator name

 required

Returns:

Type Description
type[SemanticRule] | SemanticRule | None

Validator or None if not found
Source code in src/dppvalidator/plugins/registry.py 
 95
 96
 97
 98
 99
100
101
102
103
104
def get_validator(self, name: str) -> type[SemanticRule] | SemanticRule | None:
    """Get a validator by name.

    Args:
        name: Validator name

    Returns:
        Validator or None if not found
    """
    return self._validators.get(name)

register_exporter(name, exporter)

Register an exporter plugin.

Parameters:

Name Type Description Default
name str

Unique name for the exporter

 required
exporter type[Exporter] | Exporter

Exporter class or instance implementing Exporter protocol

 required
Source code in src/dppvalidator/plugins/registry.py 
63
64
65
66
67
68
69
70
71
def register_exporter(self, name: str, exporter: type[Exporter] | Exporter) -> None:
    """Register an exporter plugin.

    Args:
        name: Unique name for the exporter
        exporter: Exporter class or instance implementing Exporter protocol
    """
    self._exporters[name] = exporter
    logger.debug("Manually registered exporter: %s", name)

register_validator(name, validator)

Register a validator plugin.

Parameters:

Name Type Description Default
name str

Unique name for the validator

 required
validator type[SemanticRule] | SemanticRule

Validator class or instance implementing SemanticRule protocol

 required
Source code in src/dppvalidator/plugins/registry.py 
53
54
55
56
57
58
59
60
61
def register_validator(self, name: str, validator: type[SemanticRule] | SemanticRule) -> None:
    """Register a validator plugin.

    Args:
        name: Unique name for the validator
        validator: Validator class or instance implementing SemanticRule protocol
    """
    self._validators[name] = validator
    logger.debug("Manually registered validator: %s", name)

run_all_validators(passport, *, strict=False)

Run all registered validator plugins.

Parameters:

Name Type Description Default
passport DigitalProductPassport

Parsed passport to validate

 required
strict bool

If True, raise PluginError on plugin failures instead of returning a warning. Useful for CI/CD pipelines.

 False

Returns:

Type Description
list[ValidationError]

List of validation errors from all plugins

Raises:

Type Description
PluginError

If strict=True and a plugin fails to execute
Source code in src/dppvalidator/plugins/registry.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def run_all_validators(
    self,
    passport: DigitalProductPassport,
    *,
    strict: bool = False,
) -> list[ValidationError]:
    """Run all registered validator plugins.

    Args:
        passport: Parsed passport to validate
        strict: If True, raise PluginError on plugin failures instead of
            returning a warning. Useful for CI/CD pipelines.

    Returns:
        List of validation errors from all plugins

    Raises:
        PluginError: If strict=True and a plugin fails to execute
    """
    errors: list[ValidationError] = []

    for name, validator in self._validators.items():
        try:
            instance = validator() if isinstance(validator, type) else validator

            if hasattr(instance, "check"):
                violations = instance.check(passport)
                rule_id = getattr(instance, "rule_id", f"PLG_{name.upper()}")
                severity = getattr(instance, "severity", "error")

                for path, message in violations:
                    errors.append(
                        ValidationError(
                            path=path,
                            message=message,
                            code=rule_id,
                            layer="plugin",
                            severity=severity,
                        )
                    )

        except (AttributeError, TypeError, ValueError, RuntimeError) as e:
            if strict:
                raise PluginError(f"Plugin '{name}' failed: {e}") from e
            logger.warning("Plugin %s failed: %s", name, e)
            errors.append(
                ValidationError(
                    path="$",
                    message=f"Plugin '{name}' failed: {e}",
                    code="PLG001",
                    layer="plugin",
                    severity="warning",
                )
            )

    return errors

unregister_exporter(name)

Unregister an exporter plugin.

Parameters:

Name Type Description Default
name str

Name of exporter to unregister

 required

Returns:

Type Description
bool

True if exporter was found and removed
Source code in src/dppvalidator/plugins/registry.py 
84
85
86
87
88
89
90
91
92
93
def unregister_exporter(self, name: str) -> bool:
    """Unregister an exporter plugin.

    Args:
        name: Name of exporter to unregister

    Returns:
        True if exporter was found and removed
    """
    return self._exporters.pop(name, None) is not None

unregister_validator(name)

Unregister a validator plugin.

Parameters:

Name Type Description Default
name str

Name of validator to unregister

 required

Returns:

Type Description
bool

True if validator was found and removed
Source code in src/dppvalidator/plugins/registry.py 
73
74
75
76
77
78
79
80
81
82
def unregister_validator(self, name: str) -> bool:
    """Unregister a validator plugin.

    Args:
        name: Name of validator to unregister

    Returns:
        True if validator was found and removed
    """
    return self._validators.pop(name, None) is not None

options: show_source: false

Plugin Discovery

Plugins are discovered via Python entry points.

Entry Points

# pyproject.toml
[project.entry-points."dppvalidator.validators"]
my_validator = "my_package:MyValidator"

[project.entry-points."dppvalidator.exporters"]
my_exporter = "my_package:MyExporter"

Usage Example

from dppvalidator.plugins import PluginRegistry
from dppvalidator.plugins.registry import get_default_registry

# Use singleton registry (recommended)
registry = get_default_registry()

# List available plugins
print("Validators:", registry.validator_names)
print("Exporters:", registry.exporter_names)

# Get a specific plugin
validator = registry.get_validator("my_validator")
exporter = registry.get_exporter("my_exporter")

# Manual registration (for testing)
registry = PluginRegistry(auto_discover=False)
registry.register_validator("custom", MyCustomValidator)

Strict Mode

For CI/CD pipelines, use strict mode to raise exceptions on plugin failures:

from dppvalidator.plugins.registry import get_default_registry, PluginError

registry = get_default_registry()

try:
    errors = registry.run_all_validators(passport, strict=True)
except PluginError as e:
    print(f"Plugin failed: {e}")
    sys.exit(1)

Creating Plugins

See the Plugin Development Guide for detailed instructions.