from collections.abc import Mapping
from pathlib import Path
from pprint import pformat
from pydantic import ValidationError
[docs]
class YamlError(Exception):
"""
Base exception for YAML errors.
"""
message_template = "An error occurred with file: {file_path}"
def __init__(self, file_path: Path, **kwargs):
self.file_path = file_path
# Format the template defined in the subclass
message = self.message_template.format(file_path=file_path, **kwargs)
super().__init__(message)
[docs]
class YamlOutputError(YamlError):
"""
Raised when the final workflow cannot be exported as a YAML file.
This would likely indicate that one of the preceding transformation steps
led to a format that is no longer able to be exported as a YAML file.
"""
message_template = "File {file_path} could not be output by ruamel-yaml."
[docs]
class YamlParsingError(YamlError):
"""
Raised when the rendered template is not a valid YAML file, as it cannot be
parsed by ruamel-yaml.
"""
message_template = (
"File {file_path} could not be parsed. Check for invalid YAML syntax."
)
[docs]
class TemplateRenderingError(YamlError):
"""
Raised when Jinja fails to modify the template. It may be that a Jinja
variable was not defined, a brace was not closed, etc.
"""
message_template = (
"File {file_path} failed to render. Check for Jinja-related errors."
)
[docs]
class InvalidWorkflowPatcherYamlError(YamlError):
"""
Raised when the :class:`WorkflowPatcher` failed the validation constraints of
:class:`WorkflowPatcherConfig`.
"""
message_template = (
"File '{file_path}' is malformed; "
"it failed Pydantic validation with {error_count} errors.\n"
"Validation issue information:\n"
"{validation_details}"
)
def __init__(self, file_path: Path, validation_error: ValidationError):
validation_details = (
f"\033[31m{pformat(validation_error.errors(), sort_dicts=False)}\033[0m"
)
self.validation_error = validation_error
super().__init__(
file_path=file_path,
error_count=validation_error.error_count(),
validation_details=validation_details,
)
[docs]
class InvalidWorkflowPatcherEntryError(YamlError):
"""
Raised when the :class:`WorkflowPatcher` is used but one of the specified keys it
listed does not exist in the relevant workflow template file.
"""
message_template = (
"In file '{file_path}', an entry '{entry}' does not exist in "
"the workflow template. Please fix the entry."
)
[docs]
class InvalidWorkflowNameError(ValueError):
"""
Raised when a workflow name is not one of the available PTB templates.
"""
def __init__(self, workflow_name: str, valid_workflows):
super().__init__(
f"Invalid workflow: {workflow_name}. Must be one of {valid_workflows}"
)
[docs]
class NotMaintainedWorkflowError(ValueError):
"""
Raised when a PTB-seeded workflow is requested in an existing project.
"""
def __init__(self, workflow_name: str):
super().__init__(
f"Workflow '{workflow_name}' is a PTB-seeded workflow that is "
"originally provided by the PTB and can only be seeded for a new project."
)
[docs]
class YamlKeyError(Exception):
"""
Base exception for when a specified value cannot be found in a YAML.
"""
message_template = "An error occurred with job: '{job_name}'"
def __init__(self, **kwargs):
# Store all attributes dynamically (job_name, step_id, etc.)
for key, value in kwargs.items():
setattr(self, key, value)
self._data = kwargs
# Format the template using the passed-in arguments
super().__init__(self.message_template.format(**kwargs))
@property
def entry(self) -> Mapping[str, str]:
return self._data
[docs]
class YamlJobValueError(YamlKeyError):
"""
Raised when a job cannot be found in a YAML file.
"""
message_template = "Job '{job_name}' could not be found"
[docs]
class YamlStepIdValueError(YamlKeyError):
"""
Raised when a step_id associated with a specific job cannot be found in a YAML file.
"""
message_template = "Step_id '{step_id}' not found in job '{job_name}'"
Exasol Toolbox