Coverage for src / lexigram / contracts / exceptions / execution.py: 0%

1"""Execution-related exception classes.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from lexigram.contracts.exceptions.base import LexigramError 

8 

9 

10class PipelineExecutionError(LexigramError): 

11 """Error raised during pipeline execution.""" 

12 

13 _code = "LEX_ERR_PIPE_001" 

14 

15 def __init__( 

16 self, 

17 step_name: str, 

18 error: Exception, 

19 details: dict[str, Any] | None = None, 

20 **kwargs: Any, 

21 ) -> None: 

22 self.step_name = step_name 

23 self.error = error 

24 message = f"Pipeline step '{step_name}' failed: {error}" 

25 super().__init__(message, details=details, **kwargs) 

26 

27 

28class PipelineStepError(LexigramError): 

29 """Error raised by a pipeline step.""" 

30 

31 _code = "LEX_ERR_PIPE_002" 

32 

33 def __init__(self, message: str = "Pipeline step error", **kwargs: Any) -> None: 

34 super().__init__(message, **kwargs) 

35 

36 

37__all__ = ["PipelineExecutionError", "PipelineStepError"]