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
Python
 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
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
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,
        schema_version: str | None = None,
    ) -> 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.
            schema_version: Resolved UNTP DPP version of the payload. When
                supplied, plugins that declare an ``applies_to_versions``
                tuple are skipped for non-matching versions (the engine's
                per-version dispatch contract — same pattern the built-in
                semantic-rule registry uses via ``ALL_RULES_BY_VERSION``).
                Plugins without that attribute keep running for every
                payload — back-compat for pre-Phase-6 plugins.

        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

                # Per-version filter. ``applies_to_versions`` is the
                # declarative version-pin contract documented in
                # ``docs/guides/plugins.md``. We honour it on both
                # the class and the instance so authors can set it
                # either way. Plugins that don't declare it run for
                # every version (back-compat).
                applies = getattr(instance, "applies_to_versions", None)
                if applies and schema_version and schema_version not in applies:
                    continue

                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
Python
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
Python
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
Python
 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
Python
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
Python
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, schema_version=None)

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
schema_version str | None

Resolved UNTP DPP version of the payload. When supplied, plugins that declare an applies_to_versions tuple are skipped for non-matching versions (the engine's per-version dispatch contract — same pattern the built-in semantic-rule registry uses via ALL_RULES_BY_VERSION). Plugins without that attribute keep running for every payload — back-compat for pre-Phase-6 plugins.

 None

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
Python
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
def run_all_validators(
    self,
    passport: DigitalProductPassport,
    *,
    strict: bool = False,
    schema_version: str | None = None,
) -> 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.
        schema_version: Resolved UNTP DPP version of the payload. When
            supplied, plugins that declare an ``applies_to_versions``
            tuple are skipped for non-matching versions (the engine's
            per-version dispatch contract — same pattern the built-in
            semantic-rule registry uses via ``ALL_RULES_BY_VERSION``).
            Plugins without that attribute keep running for every
            payload — back-compat for pre-Phase-6 plugins.

    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

            # Per-version filter. ``applies_to_versions`` is the
            # declarative version-pin contract documented in
            # ``docs/guides/plugins.md``. We honour it on both
            # the class and the instance so authors can set it
            # either way. Plugins that don't declare it run for
            # every version (back-compat).
            applies = getattr(instance, "applies_to_versions", None)
            if applies and schema_version and schema_version not in applies:
                continue

            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
Python
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
Python
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

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

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

Usage Example

Python
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:

Python
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.