"""Structured diagnostics shared across Buffalo workflows."""
from __future__ import annotations
from dataclasses import dataclass, field
from enum import StrEnum
__all__ = [
"Diagnostic",
"DiagnosticLocation",
"DiagnosticReport",
"DiagnosticSeverity",
"OperationResult",
]
[docs]
class DiagnosticSeverity(StrEnum):
"""Severity level for a structured diagnostic."""
ERROR = "error"
"Failures that should block the workflow result."
WARNING = "warning"
"Issues that preserve a usable result but require attention."
INFO = "info"
"Non-failing informational messages."
[docs]
@dataclass(frozen=True, slots=True)
class DiagnosticLocation:
"""
Optional context identifying where a diagnostic applies.
Notes
-----
All fields are optional so the same structure can support coarse
workflow-level diagnostics as well as fine-grained field or region
localization.
"""
object_path: str | None = None
"""
Optional logical object or workflow path associated with the diagnostic.
This can identify a schema object, runtime object, or workflow stage.
"""
field_path: str | None = None
"""
Optional path diagnostic for localizing object.
This is a field-level or payload-level path that localizes the diagnostic
within one object.
"""
geometry_region: str | None = None
"""
Optional geometry-oriented label.
This label provides additional information, such as ``leading_edge`` or
``trailing_edge``, when the diagnostic applies to one region of a
geometric object.
"""
[docs]
@dataclass(frozen=True, slots=True)
class Diagnostic:
"""
Structured diagnostic suitable for programmatic reporting.
Raises
------
ValueError
If ``code`` or ``message`` is an empty string.
Notes
-----
The ``code`` field is intended to remain stable enough for programmatic
filtering and test assertions.
The ``message`` field is intended for direct human consumption.
"""
severity: DiagnosticSeverity
"""
Severity classification for the diagnostic.
"""
code: str
"""
Stable machine-readable diagnostic code.
"""
message: str
"""
Human-readable message describing the issue or informational result.
"""
location: DiagnosticLocation | None = None
"""
Optional location metadata that localizes where the diagnostic applies.
"""
def __post_init__(self) -> None:
"""Validate required diagnostic fields."""
if not self.code:
msg = "Diagnostic code must be a non-empty string."
raise ValueError(msg)
if not self.message:
msg = "Diagnostic message must be a non-empty string."
raise ValueError(msg)
[docs]
@dataclass(frozen=True, slots=True)
class DiagnosticReport:
"""
Grouped diagnostics emitted by one workflow.
Notes
-----
Reports preserve input order so workflows can keep diagnostics in the
order they were discovered or produced.
Convenience properties such as :attr:`has_errors` and
:meth:`for_severity` support quick inspection without changing the stored
ordering.
"""
entries: tuple[Diagnostic, ...] = field(default_factory=tuple)
"""
Ordered diagnostics emitted while performing one operation.
"""
@property
def has_errors(self) -> bool:
"""Return whether the report contains any errors."""
return bool(self.for_severity(DiagnosticSeverity.ERROR))
@property
def has_warnings(self) -> bool:
"""Return whether the report contains any warnings."""
return bool(self.for_severity(DiagnosticSeverity.WARNING))
@property
def has_infos(self) -> bool:
"""Return whether the report contains any infos."""
return bool(self.for_severity(DiagnosticSeverity.INFO))
[docs]
def for_severity(
self,
severity: DiagnosticSeverity,
) -> tuple[Diagnostic, ...]:
"""
Return diagnostics matching one severity.
Parameters
----------
severity : DiagnosticSeverity
Severity level to filter from ``entries``.
Returns
-------
tuple[Diagnostic, ...]
Ordered diagnostics whose severity matches ``severity`` exactly.
"""
return tuple(
entry for entry in self.entries if entry.severity is severity
)
[docs]
@dataclass(frozen=True, slots=True)
class OperationResult[T]:
"""
Return one value together with structured diagnostics.
Notes
-----
This type is useful for workflows that should return a usable value while
still exposing warnings, informational messages, or recoverable issues to
the caller.
"""
value: T
"""
Primary workflow return value.
"""
diagnostics: DiagnosticReport
"""
Structured diagnostics emitted while producing ``value``.
"""
@property
def has_errors(self) -> bool:
"""Return whether attached diagnostics contain errors."""
return self.diagnostics.has_errors
@property
def has_warnings(self) -> bool:
"""Return whether attached diagnostics contain warnings."""
return self.diagnostics.has_warnings
@property
def has_infos(self) -> bool:
"""Return whether attached diagnostics contain infos."""
return self.diagnostics.has_infos