Coverage for src / lexigram / contracts / queue / errors.py: 54%

1# lexigram-contracts/src/lexigram/contracts/queue/errors.py 

2"""Queue error types.""" 

3 

4from __future__ import annotations 

5 

6from typing import Any 

7 

8from lexigram.contracts.exceptions.domain import DomainError 

9 

10 

11class QueueError(DomainError): 

12 """Base for expected, recoverable queue / bus failures. 

13 

14 Attributes: 

15 backend: Name of the queue backend (e.g. "redis", "kafka"). 

16 topic: Topic or queue name where the failure occurred. 

17 """ 

18 

19 _code = "LEX_ERR_QUEUE_001" 

20 

21 def __init__( 

22 self, 

23 message: str = "Queue error", 

24 *, 

25 backend: str = "unknown", 

26 topic: str = "unknown", 

27 **kwargs: Any, 

28 ) -> None: 

29 details = kwargs.pop("details", {}) 

30 details["backend"] = backend 

31 details["topic"] = topic 

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

33 self.backend = backend 

34 self.topic = topic 

35 

36 

37__all__ = ["QueueError"]