Coverage for src / lexigram / contracts / infra / tasks / progress.py: 0%

1"""Progress tracking protocol and data types for long-running tasks. 

2 

3Provides the cross-package contract for task progress reporting, enabling any 

4package (``lexigram-admin``, ``lexigram-web``, …) to depend on the protocol 

5without importing from ``lexigram-tasks`` directly. 

6 

7Example: 

8 ```python 

9 from lexigram.contracts.infra.tasks.progress import ( 

10 ProgressTrackerProtocol, 

11 ProgressSnapshot, 

12 ProgressStatus, 

13 ) 

14 

15 class ReportController: 

16 def __init__(self, progress: ProgressTrackerProtocol) -> None: 

17 self._progress = progress 

18 

19 async def stream(self, task_id: str) -> AsyncIterator[ProgressSnapshot]: 

20 async for snap in self._progress.subscribe(task_id): 

21 yield snap 

22 ``` 

23""" 

24 

25from __future__ import annotations 

26 

27from collections.abc import AsyncIterator 

28from dataclasses import dataclass 

29from enum import Enum 

30from typing import Protocol, runtime_checkable 

31 

32 

33class ProgressStatus(str, Enum): 

34 """Lifecycle states for a tracked task. 

35 

36 Values are lowercase strings so they round-trip cleanly through JSON. 

37 """ 

38 

39 PENDING = "pending" 

40 RUNNING = "running" 

41 COMPLETE = "complete" 

42 FAILED = "failed" 

43 

44 

45@dataclass(frozen=True) 

46class ProgressSnapshot: 

47 """Immutable point-in-time view of a task's progress. 

48 

49 Args: 

50 task_id: Unique identifier for the tracked task. 

51 current: Number of units completed so far. 

52 total: Total number of units to complete (0 means unknown). 

53 status: Current lifecycle state. 

54 message: Human-readable status message. 

55 error: Error description when ``status`` is ``FAILED``; empty otherwise. 

56 """ 

57 

58 task_id: str 

59 current: int 

60 total: int 

61 status: ProgressStatus 

62 message: str = "" 

63 error: str = "" 

64 

65 @property 

66 def percent(self) -> float: 

67 """Completion percentage in the range ``[0.0, 100.0]``. 

68 

69 Returns ``0.0`` when ``total`` is unknown (zero). 

70 """ 

71 if self.total == 0: 

72 return 0.0 

73 return min(100.0, self.current / self.total * 100) 

74 

75 

76@runtime_checkable 

77class ProgressTrackerProtocol(Protocol): 

78 """Protocol for tracking and broadcasting task progress. 

79 

80 Implementations must be safe for concurrent use — multiple coroutines 

81 may call ``update`` on the same ``task_id`` simultaneously, and multiple 

82 consumers may subscribe to the same task. 

83 

84 The ``subscribe`` method returns a live ``AsyncIterator`` that yields a 

85 new :class:`ProgressSnapshot` every time ``update``, ``complete``, or 

86 ``fail`` is called for the given task. The iterator terminates 

87 automatically when the task reaches a terminal state (``COMPLETE`` or 

88 ``FAILED``). 

89 

90 Example: 

91 ```python 

92 async def export_data(tracker: ProgressTrackerProtocol) -> None: 

93 records = await fetch_all() 

94 total = len(records) 

95 for i, record in enumerate(records, start=1): 

96 await process(record) 

97 await tracker.update("export-1", i, total, f"Row {i}/{total}") 

98 await tracker.complete("export-1", "Export finished") 

99 ``` 

100 """ 

101 

102 async def update( 

103 self, 

104 task_id: str, 

105 current: int, 

106 total: int, 

107 message: str = "", 

108 ) -> None: 

109 """Record incremental progress for a task. 

110 

111 Args: 

112 task_id: Unique identifier for the task. 

113 current: Units completed so far. 

114 total: Total units to process (0 = unknown). 

115 message: Optional human-readable status message. 

116 """ 

117 ... 

118 

119 async def complete(self, task_id: str, result: str = "") -> None: 

120 """Mark a task as successfully completed. 

121 

122 Closes all active subscriptions for ``task_id`` after broadcasting the 

123 terminal :class:`ProgressSnapshot`. 

124 

125 Args: 

126 task_id: Unique identifier for the task. 

127 result: Optional human-readable completion message. 

128 """ 

129 ... 

130 

131 async def fail(self, task_id: str, error: str) -> None: 

132 """Mark a task as failed. 

133 

134 Closes all active subscriptions for ``task_id`` after broadcasting the 

135 terminal :class:`ProgressSnapshot`. 

136 

137 Args: 

138 task_id: Unique identifier for the task. 

139 error: Description of the failure. 

140 """ 

141 ... 

142 

143 async def get(self, task_id: str) -> ProgressSnapshot | None: 

144 """Return the current progress state for a task. 

145 

146 Args: 

147 task_id: Unique identifier for the task. 

148 

149 Returns: 

150 The most recent :class:`ProgressSnapshot`, or ``None`` if the task 

151 is not known to this tracker. 

152 """ 

153 ... 

154 

155 def subscribe(self, task_id: str) -> AsyncIterator[ProgressSnapshot]: 

156 """Subscribe to live progress updates for a task. 

157 

158 Returns an async iterator that yields a new :class:`ProgressSnapshot` 

159 each time the task's state changes. The iterator stops automatically 

160 when the task reaches a terminal state (``COMPLETE`` or ``FAILED``). 

161 

162 If the task is already in a terminal state when ``subscribe`` is 

163 called, the iterator yields the final snapshot once and then stops. 

164 

165 Args: 

166 task_id: Unique identifier for the task to observe. 

167 

168 Returns: 

169 An :class:`AsyncIterator` of :class:`ProgressSnapshot` objects. 

170 

171 Example: 

172 ```python 

173 async for snap in tracker.subscribe("job-42"): 

174 print(f"{snap.percent:.1f}% — {snap.message}") 

175 ``` 

176 """ 

177 ... 

178 

179 

180__all__ = [ 

181 "ProgressSnapshot", 

182 "ProgressStatus", 

183 "ProgressTrackerProtocol", 

184]