Coverage for /home/runner/work/bijux-cli/bijux-cli/src/bijux_cli/contracts/retry.py: 100%

1# SPDX-License-Identifier: MIT 

2# Copyright © 2025 Bijan Mousavi 

3 

4"""Defines the contract for operation retry policies. 

5 

6This module specifies the `RetryPolicyProtocol`, a formal interface that any 

7service providing retry logic for potentially failing operations must 

8implement. This is particularly useful for handling transient errors in 

9network requests or other I/O-bound tasks. 

10""" 

11 

12from __future__ import annotations 

13 

14from collections.abc import Awaitable, Callable 

15from typing import Protocol, TypeVar, runtime_checkable 

16 

17T = TypeVar("T") 

18 

19 

20@runtime_checkable 

21class RetryPolicyProtocol(Protocol): 

22 """Defines the contract for retry policies in asynchronous operations. 

23 

24 This interface specifies the methods for executing an operation with retry 

25 logic (e.g., exponential backoff) and for resetting the policy's internal 

26 state. 

27 """ 

28 

29 async def run( 

30 self, supplier: Callable[[], Awaitable[T]], seconds: float = 1.0 

31 ) -> T: 

32 """Runs an asynchronous operation with a retry policy. 

33 

34 Implementations of this method will repeatedly call the `supplier` 

35 until it succeeds or the retry policy is exhausted. 

36 

37 Args: 

38 supplier (Callable[[], Awaitable[T]]): A no-argument function that 

39 returns an awaitable (e.g., a coroutine). 

40 seconds (float): The timeout for each attempt in seconds. 

41 

42 Returns: 

43 T: The successful result of the operation. 

44 """ 

45 ... 

46 

47 def reset(self) -> None: 

48 """Resets the internal state of the retry policy. 

49 

50 This is useful for reusing a policy instance for a new, independent 

51 operation. 

52 """ 

53 ...