# Copyright 2025–2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""Define exceptions and validation issues used throughout MAFw.
:Author: Bulgheroni Antonio
:Description: Provide common error types for framework and steering failures.
"""
[docs]
class MAFwException(Exception):
"""Base class for MAFwException"""
pass
[docs]
class ProcessorParameterError(MAFwException):
"""Error with a processor parameter"""
pass
[docs]
class InvalidConfigurationError(MAFwException):
"""Error with the configuration of a processor"""
pass
[docs]
class MissingOverloadedMethod(UserWarning):
"""
Warning issued when the user did not overload a required method.
It is a warning and not an error because the execution framework might still work, but the results might be
different from what is expected.
"""
pass
[docs]
class MissingSuperCall(UserWarning):
"""
Warning issued when the user did not invoke the super method for some specific processor methods.
Those methods (like :meth:`~mafw.processor.Processor.start` and :meth:`~mafw.processor.Processor.finish`) have not
empty implementation also in the base class, meaning that if the user forgets to call *super* in their overloads,
then the basic implementation will be gone.
It is a warning and not an error because the execution framework might sill work, but the results might be
different from what is expected.
"""
pass
[docs]
class SteeringFileDeprecation(UserWarning):
"""
Warning issued when a legacy deprecated configuration is used in the steering file.
It is a warning and not an error because the execution framework might still work, but the configuration
may be removed in future versions.
"""
pass
[docs]
class AbortProcessorException(MAFwException):
"""Exception raised during the execution of a processor requiring immediate exit."""
pass
[docs]
class RunnerNotInitialized(MAFwException):
"""Exception raised when attempting to run a not initialized Runner."""
pass
[docs]
class InvalidSteeringFile(MAFwException):
"""Exception raised when validating an invalid steering file"""
pass
[docs]
class ValidationIssue(InvalidSteeringFile):
"""Base class for steering validation issues."""
def __init__(self, message: str) -> None:
super().__init__(message)
[docs]
class MissingProcessorsToRunIssue(ValidationIssue):
"""Issue raised when the steering file does not declare ``processors_to_run``."""
def __init__(self) -> None:
super().__init__('Missing processors_to_run')
[docs]
class ProcessorsToRunNotListIssue(ValidationIssue):
"""Issue raised when ``processors_to_run`` is not a list."""
def __init__(self) -> None:
super().__init__('processors_to_run must be a list')
[docs]
class InvalidStorageConfigurationIssue(ValidationIssue):
"""Issue raised when a storage configuration value is invalid.
:param detail: Description of the invalid section, volume, or property.
.. versionadded:: 2.4
"""
def __init__(self, detail: str) -> None:
super().__init__(f'Invalid StorageConfiguration: {detail}')
[docs]
class MissingUserInterfaceSectionIssue(ValidationIssue):
"""Issue raised when the ``UserInterface`` section is missing."""
def __init__(self) -> None:
super().__init__('Missing UserInterface section')
[docs]
class InvalidEnvironmentFilesIssue(ValidationIssue):
"""Issue raised when ``env_files`` is not a list of strings."""
def __init__(self) -> None:
super().__init__("'env_files' must be a list of strings")
[docs]
class UnknownProcessorsToRunEntryIssue(ValidationIssue):
"""Issue raised when a processor reference in ``processors_to_run`` cannot be resolved."""
def __init__(self, entry: str) -> None:
super().__init__(f'Processor reference {entry} not defined')
[docs]
class UnknownGroupMemberIssue(ValidationIssue):
"""Issue raised when a group references an unknown processor or nested group."""
def __init__(self, group: str, member: str) -> None:
super().__init__(f"Group '{group}' references unknown processor or group '{member}'")
[docs]
class DuplicateReplicaIssue(ValidationIssue):
"""Issue raised when a processor declares duplicate replica identifiers."""
def __init__(self, base: str, replica: str) -> None:
super().__init__(f"Duplicate replica identifier '{replica}' for processor '{base}'")
[docs]
class CyclicGroupIssue(ValidationIssue):
"""Issue raised when cyclic dependencies exist within processor groups."""
def __init__(self, node: str) -> None:
super().__init__(f"Cyclic group dependency detected at '{node}'")
[docs]
class InvalidFilterConditionIssue(ValidationIssue):
"""Issue raised when a filter condition is incomplete or invalid."""
def __init__(
self,
processor: str,
model: str,
condition: str,
*,
field_name: str | None = None,
reason: str | None = None,
) -> None:
descriptor = f"Condition '{condition}' for model '{model}' in processor '{processor}'"
if field_name:
descriptor += f" field '{field_name}'"
descriptor += ' is invalid'
if reason:
descriptor += f': {reason}'
super().__init__(descriptor)
[docs]
class InvalidFilterLogicIssue(ValidationIssue):
"""Issue raised when a filter logic expression is invalid."""
def __init__(self, processor: str, target: str, *, detail: str | None = None) -> None:
descriptor = f"Logic expression for {target} on processor '{processor}' is invalid"
if detail:
descriptor += f': {detail}'
super().__init__(descriptor)
[docs]
class UnknownProcessor(MAFwException):
"""Exception raised when an attempt is made to create an unknown processor"""
pass
[docs]
class AmbiguousProcessor(MAFwException):
"""
Exception raised when an unqualified processor name matches two or more loaded processors.
Two installed plugin packages may export processors that share the same short class name
(the ``Short_Name``). In that case an unqualified reference such as ``SameNameProcessor`` is
*ambiguous*: the name is known, but it is under-specified. Rather than silently picking one of
the colliding processors, MAFw refuses the reference and reports the available
``Qualified_Name`` alternatives (``package.ShortName``), one per line, so the user can select the
intended processor unambiguously.
.. versionadded:: v2.4
:param short_name: The ambiguous short name that matched several processors.
:type short_name: str
:param alternatives: The sorted qualified names available for that short name.
:type alternatives: list[str]
"""
def __init__(self, short_name: str, alternatives: list[str]) -> None:
# Build a descriptive, multi-line message listing the qualified alternatives so that the
# scientist user can copy the exact qualified name they intend to run.
lines = '\n'.join(f' - {alternative}' for alternative in alternatives)
message = (
f"Processor name '{short_name}' is ambiguous; it is provided by multiple packages. "
f'Use one of the qualified names:\n{lines}'
)
super().__init__(message)
[docs]
class UnknownProcessorGroup(MAFwException):
"""Exception raised when an attempt is made to create an unknown processor group"""
pass
[docs]
class UnknownDBEngine(MAFwException):
"""Exception raised when the user provided an unknown db engine"""
pass
[docs]
class MissingDatabase(MAFwException):
"""Exception raised when a processor requiring a database connection is being operated without a database"""
[docs]
class MissingAttribute(MAFwException):
"""Exception raised when an attempt is made to execute a statement without a required parameter/attributes"""
[docs]
class ParserConfigurationError(MAFwException):
"""Exception raised when an error occurred during the configuration of a filename parser"""
[docs]
class ParsingError(MAFwException):
"""Exception raised when a regular expression parsing failed"""
[docs]
class MissingSQLStatement(MAFwException):
"""Exception raised when a Trigger is created without any SQL statements."""
[docs]
class UnsupportedDatabaseError(Exception):
"""Error raised when a feature is not supported by the database."""
[docs]
class ModelError(MAFwException):
"""Exception raised when an error in a DB Model class occurs"""
[docs]
class MissingOptionalDependency(UserWarning):
"""UserWarning raised when an optional dependency is required"""
[docs]
class PlotterMixinNotInitialized(MAFwException):
"""Exception raised when a plotter mixin has not properly initialized"""