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

1"""Task and job queue protocol class definitions.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6from collections.abc import Awaitable 

7from typing import TYPE_CHECKING, Any, Protocol, TypeVar, runtime_checkable 

8 

9if TYPE_CHECKING: 

10 from lexigram.contracts.core import HealthCheckResult 

11 from lexigram.contracts.core.result import Result 

12 from lexigram.contracts.infra.tasks.enums import JobStatus 

13 from lexigram.contracts.infra.tasks.exceptions import TaskQueueError 

14 

15T = TypeVar("T") 

16 

17 

18@runtime_checkable 

19class TaskManagerProtocol(Protocol): 

20 """Shared background-task management service. 

21 

22 Implemented by ``lexigram.tasks.background_task_manager.BackgroundTaskManager`` 

23 (LEX-006). Consumers resolve this protocol from the container instead of 

24 importing the implementation package (e.g. ``lexigram.monitor``). 

25 """ 

26 

27 def track(self, coro: Awaitable[T]) -> asyncio.Task[T]: 

28 """Track a coroutine as a background task.""" 

29 

30 def track_named(self, name: str, coro: Awaitable[T]) -> asyncio.Task[T]: 

31 """Track a coroutine as a named background task.""" 

32 

33 @property 

34 def pending_count(self) -> int: 

35 """Return the number of tracked, unfinished tasks.""" 

36 

37 async def shutdown(self, timeout: float = 30.0) -> None: 

38 """Cancel all tracked tasks and wait for completion within *timeout*.""" 

39 

40 

41@runtime_checkable 

42class JobProtocol(Protocol): 

43 """Protocol for JobProtocol objects.""" 

44 

45 id: str 

46 name: str 

47 args: tuple[Any, ...] 

48 kwargs: dict[str, Any] 

49 priority: int 

50 status: JobStatus 

51 max_retries: int 

52 timeout: float | None 

53 correlation_id: str | None 

54 causation_id: str | None 

55 

56 

57@runtime_checkable 

58class TaskQueueProtocol(Protocol): 

59 """Protocol for task queue implementations. 

60 

61 This defines the contract for background task queues 

62 (Redis-based, PostgreSQL-based, in-memory, etc.). 

63 

64 Example: 

65 ```python 

66 class RedisTaskQueue: 

67 async def enqueue(self, task: Task) -> str: 

68 task_id = str(uuid4()) 

69 await self._redis.rpush("tasks", task.to_json()) 

70 return task_id 

71 

72 async def dequeue(self) -> Task | None: 

73 data = await self._redis.lpop("tasks") 

74 return Task.from_json(data) if data else None 

75 ``` 

76 """ 

77 

78 async def enqueue(self, task: Any) -> Result[str, TaskQueueError]: 

79 """Add a task to the queue. 

80 

81 Args: 

82 task: Task instance to enqueue. 

83 

84 Returns: 

85 Ok(task_id) on success; Err(TaskQueueError) if the queue reports 

86 a recoverable failure (e.g. full queue, invalid payload). Only 

87 infrastructure failures (lost connection) are raised as exceptions. 

88 """ 

89 ... 

90 

91 async def dequeue(self) -> Any | None: 

92 """Remove and return the next task. 

93 

94 Returns: 

95 Next Task or None if queue is empty. 

96 """ 

97 ... 

98 

99 async def get_task_count(self) -> int: 

100 """Get the number of tasks in the queue. 

101 

102 Returns: 

103 Number of pending tasks. 

104 """ 

105 ... 

106 

107 async def clear(self) -> None: 

108 """Clear all tasks from the queue.""" 

109 ... 

110 

111 async def get_job(self, job_id: str) -> Any | None: 

112 """Get a job by ID. 

113 

114 Args: 

115 job_id: Unique job identifier. 

116 

117 Returns: 

118 Job data or None if not found. 

119 """ 

120 ... 

121 

122 async def delete_job(self, job_id: str) -> bool: 

123 """Delete a job from the queue. 

124 

125 Args: 

126 job_id: Unique job identifier. 

127 

128 Returns: 

129 True if the job was deleted. 

130 """ 

131 ... 

132 

133 async def count(self, queue_name: str, status: Any | None = None) -> int: 

134 """Count queued jobs. 

135 

136 Args: 

137 queue_name: Queue to inspect. 

138 status: Optional status filter. 

139 

140 Returns: 

141 Number of matching jobs. 

142 """ 

143 ... 

144 

145 async def ack(self, task_id: str) -> None: 

146 """Acknowledge successful processing of a dequeued task. 

147 

148 Signals that the task has been processed successfully and can 

149 be permanently discarded from any in-flight tracking. 

150 

151 Args: 

152 task_id: ID of the task to acknowledge. 

153 """ 

154 ... 

155 

156 async def nack(self, task_id: str, requeue: bool = True) -> None: 

157 """Negative-acknowledge a dequeued task. 

158 

159 Signals that the task could not be processed. The task is 

160 optionally requeued for another processing attempt. 

161 

162 Args: 

163 task_id: ID of the task to negative-acknowledge. 

164 requeue: If True, return the task to the queue for retry. 

165 If False, discard the task permanently. 

166 """ 

167 ... 

168 

169 async def close(self) -> None: 

170 """Close the queue connection and cleanup resources.""" 

171 ... 

172 

173 

174@runtime_checkable 

175class TaskExecutorProtocol(Protocol): 

176 """Protocol for task executor implementations. 

177 

178 Executors process tasks from a queue by dispatching 

179 them to registered handlers. 

180 

181 Example: 

182 ```python 

183 class TaskExecutorProtocol: 

184 async def execute(self, task: Task) -> Any: 

185 handler = self._handlers.get(task.name) 

186 if handler: 

187 return await handler(task.payload) 

188 raise UnknownTaskError(task.name) 

189 ``` 

190 """ 

191 

192 async def execute(self, task: Any) -> Any: 

193 """Execute a task using the registered handler. 

194 

195 Args: 

196 task: Task to execute. 

197 

198 Returns: 

199 Task execution result. 

200 

201 Raises: 

202 UnknownTaskError: If no handler registered for task. 

203 """ 

204 ... 

205 

206 def register_handler(self, task_name: str, handler: Any) -> None: 

207 """Register a handler for a task type. 

208 

209 Args: 

210 task_name: Name of the task type. 

211 handler: Async callable to handle the task. 

212 """ 

213 ... 

214 

215 def get_handler(self, task_name: str) -> Any | None: 

216 """Get the handler for a task type. 

217 

218 Args: 

219 task_name: Name of the task type. 

220 

221 Returns: 

222 Handler if registered, None otherwise. 

223 """ 

224 ... 

225 

226 

227@runtime_checkable 

228class TaskProviderProtocol(Protocol): 

229 """Protocol for task provider implementations. 

230 

231 Task providers integrate task processing with the framework, 

232 providing dependency injection, lifecycle management, and health monitoring. 

233 """ 

234 

235 async def register(self, container: Any) -> None: 

236 """Register task services with the DI container. 

237 

238 Args: 

239 container: DI container to register services in. 

240 """ 

241 ... 

242 

243 async def boot(self, container: Any | None = None) -> None: 

244 """Start the task provider. 

245 

246 Called by the framework on application startup. 

247 

248 Args: 

249 container: Optional DI container. 

250 """ 

251 ... 

252 

253 async def shutdown(self, app: Any) -> None: 

254 """Shutdown the task provider gracefully. 

255 

256 Called by the framework on application shutdown. 

257 

258 Args: 

259 app: Application instance. 

260 """ 

261 ... 

262 

263 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult: 

264 """Check task provider health. 

265 

266 Returns: 

267 HealthCheckResult with current status and metrics. 

268 """ 

269 ... 

270 

271 def register_handler(self, task_name: str, handler: Any) -> None: 

272 """Register a task handler. 

273 

274 Args: 

275 task_name: Name of the task type. 

276 handler: Async handler function. 

277 """ 

278 ... 

279 

280 def register_scheduled_task(self, task_func: Any) -> None: 

281 """Register a decorated task function for scheduling. 

282 

283 Args: 

284 task_func: Task function decorated with @scheduled. 

285 """ 

286 ... 

287 

288 async def enqueue_job(self, job: Any) -> str: 

289 """Enqueue a job for processing. 

290 

291 Args: 

292 job: JobProtocol instance to enqueue. 

293 

294 Returns: 

295 JobProtocol ID. 

296 """ 

297 ... 

298 

299 def schedule_job( 

300 self, 

301 job_template: Any, 

302 cron_expression: str, 

303 job_id: str | None = None, 

304 ) -> str | None: 

305 """Schedule a job with cron expression. 

306 

307 Args: 

308 job_template: JobProtocol or JobTemplateProtocol instance. 

309 cron_expression: Cron expression for scheduling. 

310 job_id: Optional job ID. 

311 

312 Returns: 

313 Scheduled job ID if successful, None otherwise. 

314 """ 

315 ... 

316 

317 def unschedule_job(self, job_id: str) -> bool: 

318 """Remove a scheduled job. 

319 

320 Args: 

321 job_id: JobProtocol ID to unschedule. 

322 

323 Returns: 

324 True if job was unscheduled, False otherwise. 

325 """ 

326 ... 

327 

328 def get_worker_stats(self) -> dict[str, Any] | None: 

329 """Get worker pool statistics. 

330 

331 Returns: 

332 Dictionary with worker statistics or None. 

333 """ 

334 ... 

335 

336 def get_scheduled_jobs(self) -> dict[str, Any] | None: 

337 """Get scheduled jobs information. 

338 

339 Returns: 

340 Dictionary with scheduled jobs info or None. 

341 """ 

342 ... 

343 

344 

345@runtime_checkable 

346class JobTemplateProtocol(Protocol): 

347 """Protocol for JobTemplateProtocol objects.""" 

348 

349 name: str 

350 args: tuple[Any, ...] 

351 kwargs: dict[str, Any] 

352 priority: int 

353 max_retries: int 

354 timeout: float | None 

355 depends_on: list[str] 

356 

357 

358@runtime_checkable 

359class TaskWorkerProtocol(Protocol): 

360 """Protocol for task worker implementations. 

361 

362 Workers consume tasks from a queue and execute them. 

363 """ 

364 

365 def __init__( 

366 self, 

367 worker_id: str, 

368 queue: Any, 

369 handler_registry: dict[str, Any], 

370 ) -> None: 

371 """Initialize worker.""" 

372 ... 

373 

374 async def start(self) -> None: 

375 """Start the worker.""" 

376 ... 

377 

378 async def stop(self) -> None: 

379 """Stop the worker.""" 

380 ... 

381 

382 

383@runtime_checkable 

384class DLQProtocol(Protocol): 

385 """Protocol for Dead Letter Queue implementations.""" 

386 

387 async def add( 

388 self, 

389 message_id: str, 

390 payload: Any, 

391 reason: str, 

392 metadata: dict[str, Any] | None = None, 

393 ) -> None: 

394 """Add a failed message to the DLQ.""" 

395 ... 

396 

397 async def get(self, message_id: str) -> dict[str, Any] | None: 

398 """Retrieve a failed message by ID.""" 

399 ... 

400 

401 async def retry(self, message_id: str) -> bool: 

402 """Move a failed message back to the main queue for reprocessing.""" 

403 ... 

404 

405 async def purge(self, message_id: str) -> bool: 

406 """Permanently delete a failed message from the DLQ.""" 

407 ... 

408 

409 async def clear(self) -> int: 

410 """Remove all failed messages and return the count removed.""" 

411 ... 

412 

413 async def size(self) -> int: 

414 """Return the number of messages currently in the DLQ.""" 

415 ... 

416 

417 async def list_failed( 

418 self, 

419 limit: int = 100, 

420 offset: int = 0, 

421 ) -> list[dict[str, Any]]: 

422 """List failed messages with pagination.""" 

423 ... 

424 

425 

426__all__ = [ 

427 "DLQProtocol", 

428 "JobProtocol", 

429 "JobTemplateProtocol", 

430 "TaskExecutorProtocol", 

431 "TaskProviderProtocol", 

432 "TaskQueueProtocol", 

433 "TaskWorkerProtocol", 

434]