mirror of
https://github.com/ivuorinen/actions.git
synced 2026-01-26 11:34:00 +00:00
* docs: update documentation * feat: validate-inputs has it's own pyproject * security: mask DOCKERHUB_PASSWORD * chore: add tokens, checkout, recrete docs, integration tests * fix: add `statuses: write` permission to pr-lint
235 lines
7.7 KiB
Python
235 lines
7.7 KiB
Python
"""Validator registry for dynamic validator discovery and loading.
|
|
|
|
Manages the registration and instantiation of validators.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
import importlib.util
|
|
import logging
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING
|
|
|
|
from .convention_mapper import ConventionMapper
|
|
from .conventions import ConventionBasedValidator
|
|
|
|
if TYPE_CHECKING:
|
|
from .base import BaseValidator
|
|
|
|
|
|
class ValidatorRegistry:
|
|
"""Registry for managing and discovering validators.
|
|
|
|
Provides dynamic loading of custom validators and fallback to convention-based validation.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
"""Initialize the validator registry."""
|
|
self._validators: dict[str, type[BaseValidator]] = {}
|
|
self._validator_instances: dict[str, BaseValidator] = {}
|
|
self._convention_mapper = ConventionMapper()
|
|
|
|
def register(self, action_type: str, validator_class: type[BaseValidator]) -> None:
|
|
"""Register a validator class for an action type.
|
|
|
|
Args:
|
|
action_type: The action type identifier
|
|
validator_class: The validator class to register
|
|
"""
|
|
self._validators[action_type] = validator_class
|
|
|
|
def register_validator(self, action_type: str, validator_class: type[BaseValidator]) -> None:
|
|
"""Register a validator class for an action type (alias for register).
|
|
|
|
Args:
|
|
action_type: The action type identifier
|
|
validator_class: The validator class to register
|
|
"""
|
|
self.register(action_type, validator_class)
|
|
# Also create and cache an instance
|
|
validator_instance = validator_class(action_type)
|
|
self._validator_instances[action_type] = validator_instance
|
|
|
|
def get_validator(self, action_type: str) -> BaseValidator:
|
|
"""Get a validator instance for the given action type.
|
|
|
|
First attempts to load a custom validator from the action directory,
|
|
then falls back to convention-based validation.
|
|
|
|
Args:
|
|
action_type: The action type identifier
|
|
|
|
Returns:
|
|
A validator instance for the action
|
|
"""
|
|
# Check cache first
|
|
if action_type in self._validator_instances:
|
|
return self._validator_instances[action_type]
|
|
|
|
# Try to load custom validator
|
|
validator = self._load_custom_validator(action_type)
|
|
|
|
# Fall back to convention-based validator
|
|
if not validator:
|
|
validator = self._load_convention_validator(action_type)
|
|
|
|
# Cache and return
|
|
self._validator_instances[action_type] = validator
|
|
return validator
|
|
|
|
def _load_custom_validator(self, action_type: str) -> BaseValidator | None:
|
|
"""Attempt to load a custom validator from the action directory.
|
|
|
|
Args:
|
|
action_type: The action type identifier
|
|
|
|
Returns:
|
|
Custom validator instance or None if not found
|
|
"""
|
|
# Convert action_type to directory name (e.g., sync_labels -> sync-labels)
|
|
action_dir = action_type.replace("_", "-")
|
|
|
|
# Look for CustomValidator.py in the action directory
|
|
project_root = Path(__file__).parent.parent.parent
|
|
custom_validator_path = project_root / action_dir / "CustomValidator.py"
|
|
|
|
if not custom_validator_path.exists():
|
|
return None
|
|
|
|
try:
|
|
# Load the module dynamically
|
|
spec = importlib.util.spec_from_file_location(
|
|
f"{action_type}_custom_validator",
|
|
custom_validator_path,
|
|
)
|
|
if not spec or not spec.loader:
|
|
return None
|
|
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
|
|
# Get the CustomValidator class
|
|
if hasattr(module, "CustomValidator"):
|
|
validator_class = module.CustomValidator
|
|
return validator_class(action_type)
|
|
|
|
except (ImportError, AttributeError, TypeError, ValueError) as e:
|
|
# Log at debug level - custom validators are optional
|
|
# Catch common errors during dynamic module loading:
|
|
# - ImportError: Module dependencies not found
|
|
# - AttributeError: Module doesn't have CustomValidator
|
|
# - TypeError: Validator instantiation failed
|
|
# - ValueError: Invalid validator configuration
|
|
logger = logging.getLogger(__name__)
|
|
logger.debug("Could not load custom validator for %s: %s", action_type, e)
|
|
|
|
return None
|
|
|
|
def _load_convention_validator(self, action_type: str) -> BaseValidator:
|
|
"""Load a convention-based validator for the action type.
|
|
|
|
Args:
|
|
action_type: The action type identifier
|
|
|
|
Returns:
|
|
Convention-based validator instance
|
|
"""
|
|
return ConventionBasedValidator(action_type)
|
|
|
|
def clear_cache(self) -> None:
|
|
"""Clear the validator instance cache."""
|
|
self._validator_instances.clear()
|
|
|
|
def list_registered(self) -> list[str]:
|
|
"""List all registered action types.
|
|
|
|
Returns:
|
|
List of registered action type identifiers
|
|
"""
|
|
return list(self._validators.keys())
|
|
|
|
def is_registered(self, action_type: str) -> bool:
|
|
"""Check if an action type has a registered validator.
|
|
|
|
Args:
|
|
action_type: The action type identifier
|
|
|
|
Returns:
|
|
True if a validator is registered, False otherwise
|
|
"""
|
|
return action_type in self._validators
|
|
|
|
def get_validator_by_type(self, validator_type: str) -> BaseValidator | None:
|
|
"""Get a validator instance by its type name.
|
|
|
|
Args:
|
|
validator_type: The validator type name (e.g., 'BooleanValidator', 'TokenValidator')
|
|
|
|
Returns:
|
|
A validator instance or None if not found
|
|
"""
|
|
# Map of validator type names to modules
|
|
validator_modules = {
|
|
"BooleanValidator": "boolean",
|
|
"CodeQLValidator": "codeql",
|
|
"DockerValidator": "docker",
|
|
"FileValidator": "file",
|
|
"NetworkValidator": "network",
|
|
"NumericValidator": "numeric",
|
|
"SecurityValidator": "security",
|
|
"TokenValidator": "token",
|
|
"VersionValidator": "version",
|
|
}
|
|
|
|
module_name = validator_modules.get(validator_type)
|
|
if not module_name:
|
|
return None
|
|
|
|
try:
|
|
# Import the module
|
|
module = importlib.import_module(f"validators.{module_name}")
|
|
# Get the validator class
|
|
validator_class = getattr(module, validator_type, None)
|
|
if validator_class:
|
|
# Create an instance with a dummy action type
|
|
return validator_class("temp")
|
|
except (ImportError, AttributeError):
|
|
# Silently ignore if custom validator module doesn't exist or class not found
|
|
pass
|
|
|
|
return None
|
|
|
|
|
|
# Global registry instance
|
|
_registry = ValidatorRegistry()
|
|
|
|
|
|
def get_validator(action_type: str) -> BaseValidator:
|
|
"""Get a validator for the given action type.
|
|
|
|
Args:
|
|
action_type: The action type identifier
|
|
|
|
Returns:
|
|
A validator instance for the action
|
|
"""
|
|
return _registry.get_validator(action_type)
|
|
|
|
|
|
def register_validator(action_type: str, validator_class: type[BaseValidator]) -> None:
|
|
"""Register a validator class for an action type.
|
|
|
|
Args:
|
|
action_type: The action type identifier
|
|
validator_class: The validator class to register
|
|
"""
|
|
_registry.register(action_type, validator_class)
|
|
|
|
|
|
def clear_cache() -> None:
|
|
"""Clear the global validator cache."""
|
|
_registry.clear_cache()
|