Skip to content

harborapi.retry

RETRY_ERRORS = (TimeoutException, NetworkError) module-attribute

P = ParamSpec('P') module-attribute

T = TypeVar('T') module-attribute

RetrySettings

Bases: BaseModel

Source code in harborapi/retry.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
class RetrySettings(BaseModel):
    enabled: bool = Field(True, description="Whether to retry requests.")
    # Required argument for backoff.on_exception
    exception: Union[Type[Exception], Tuple[Type[Exception], ...]] = Field(
        RETRY_ERRORS,
        description="Exception(s) to catch and retry on.",
    )

    # Optional arguments for backoff.on_exception
    max_tries: Optional[int] = Field(
        default=None,
        gt=0,
        description="Maximum number of tries before giving up.",
    )
    max_time: Optional[float] = Field(
        default=60,
        ge=0,
        description="Maximum number of seconds to retry for.",
    )
    wait_gen: _WaitGenerator = Field(
        default=backoff.expo,
        description="Function that generates wait times.",
    )
    jitter: Union[_Jitterer, None] = Field(
        default=backoff.full_jitter,
        description="Function that jitters wait times.",
    )
    giveup: _Predicate[Exception] = Field(
        default=DEFAULT_PREDICATE,
        description="Predicate function that determines if we should give up.",
    )
    on_success: Union[_Handler, Iterable[_Handler], None] = Field(
        default=None,
        description="Function(s) to call on success.",
    )
    on_backoff: Union[_Handler, Iterable[_Handler], None] = Field(
        default=None,
        description="Function(s) to call when backing off.",
    )
    on_giveup: Union[_Handler, Iterable[_Handler], None] = Field(
        default=None,
        description="Function(s) to call when giving up.",
    )
    raise_on_giveup: bool = Field(
        default=True,
        description="Whether to raise the exception when giving up.",
    )
    model_config = ConfigDict(extra="allow", validate_assignment=True)

    @property
    def wait_gen_kwargs(self) -> Dict[str, Any]:
        """Dict of extra model fields."""
        fields = self.model_fields.keys()
        return {
            key: value for key, value in self.model_dump().items() if key not in fields
        }

enabled = Field(True, description='Whether to retry requests.') class-attribute instance-attribute

exception = Field(RETRY_ERRORS, description='Exception(s) to catch and retry on.') class-attribute instance-attribute

max_tries = Field(default=None, gt=0, description='Maximum number of tries before giving up.') class-attribute instance-attribute

max_time = Field(default=60, ge=0, description='Maximum number of seconds to retry for.') class-attribute instance-attribute

wait_gen = Field(default=backoff.expo, description='Function that generates wait times.') class-attribute instance-attribute

jitter = Field(default=backoff.full_jitter, description='Function that jitters wait times.') class-attribute instance-attribute

giveup = Field(default=DEFAULT_PREDICATE, description='Predicate function that determines if we should give up.') class-attribute instance-attribute

on_success = Field(default=None, description='Function(s) to call on success.') class-attribute instance-attribute

on_backoff = Field(default=None, description='Function(s) to call when backing off.') class-attribute instance-attribute

on_giveup = Field(default=None, description='Function(s) to call when giving up.') class-attribute instance-attribute

raise_on_giveup = Field(default=True, description='Whether to raise the exception when giving up.') class-attribute instance-attribute

model_config = ConfigDict(extra='allow', validate_assignment=True) class-attribute instance-attribute

wait_gen_kwargs property

Dict of extra model fields.

DEFAULT_PREDICATE(e)

Predicate function that always returns False.

Source code in harborapi/retry.py 
37
38
39
def DEFAULT_PREDICATE(e: Exception) -> bool:
    """Predicate function that always returns False."""
    return False

get_backoff_kwargs(client)

Source code in harborapi/retry.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def get_backoff_kwargs(client: "HarborAsyncClient") -> Dict[str, Any]:
    retry_settings = client.retry

    # We should never get here, but just in case...
    assert retry_settings is not None, "Client has no retry settings."

    # Ignore RetrySettings.enabled, since we're already here.
    # Callers should have checked that already.

    return dict(
        exception=retry_settings.exception,
        max_tries=retry_settings.max_tries,
        max_time=retry_settings.max_time,
        wait_gen=retry_settings.wait_gen,
        jitter=retry_settings.jitter,
        giveup=retry_settings.giveup,
        on_success=retry_settings.on_success,
        on_backoff=retry_settings.on_backoff,
        on_giveup=retry_settings.on_giveup,
        raise_on_giveup=retry_settings.raise_on_giveup,
        # extra model fields become **wait_gen_kwargs
        **retry_settings.wait_gen_kwargs,
    )

retry()

Adds retry functionality to a HarborAsyncClient method.

NOTE: will fail if applied to any other class than HarborAsyncClient.

Source code in harborapi/retry.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def retry() -> Callable[[Callable[P, T]], Callable[P, T]]:
    """Adds retry functionality to a HarborAsyncClient method.

    NOTE: will fail if applied to any other class than HarborAsyncClient.
    """

    def decorator(func: Callable[P, T]) -> Callable[P, T]:
        @functools.wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
            if not args or not isinstance(args[0], _get_client_type()):
                raise TypeError(
                    "retry decorator must be applied on a HarborAsyncClient method."
                )
            client = args[0]
            if not client.retry or not client.retry.enabled:
                return func(*args, **kwargs)

            return backoff.on_exception(**get_backoff_kwargs(client))(func)(
                *args, **kwargs
            )

        return wrapper

    return decorator

Backoff types

The following types come from the backoff package. They are documented here for convenience.

_CallableT = TypeVar('_CallableT', bound=Callable[..., Any]) module-attribute

_Handler = Union[Callable[[Details], None], Callable[[Details], Coroutine[Any, Any, None]]] module-attribute

_Jitterer = Callable[[float], float] module-attribute

_MaybeCallable = Union[T, Callable[[], T]] module-attribute

_MaybeLogger = Union[str, logging.Logger, None] module-attribute

_MaybeSequence = Union[T, Sequence[T]] module-attribute

_WaitGenerator = Callable[..., Generator[float, None, None]] module-attribute

_Details

Bases: TypedDict

target instance-attribute

args instance-attribute

kwargs instance-attribute

tries instance-attribute

elapsed instance-attribute

Details

Bases: _Details

wait instance-attribute

value instance-attribute