1from typing import TYPE_CHECKING
2
3from ..config import Config
4from .irresource import IRResource
5
6if TYPE_CHECKING:
7 from .ir import IR # pragma: no cover
8
9
10class IRRetryPolicy(IRResource):
11 def __init__(
12 self,
13 ir: "IR",
14 aconf: Config,
15 rkey: str = "ir.retrypolicy",
16 kind: str = "IRRetryPolicy",
17 name: str = "ir.retrypolicy",
18 **kwargs
19 ) -> None:
20 # print("IRRetryPolicy __init__ (%s %s %s)" % (kind, name, kwargs))
21
22 super().__init__(ir=ir, aconf=aconf, rkey=rkey, kind=kind, name=name, **kwargs)
23
24 def setup(self, ir: "IR", aconf: Config) -> bool:
25 if not self.validate_retry_policy():
26 self.post_error("Invalid retry policy specified: {}".format(self))
27 return False
28
29 return True
30
31 def validate_retry_policy(self) -> bool:
32 retry_on = self.get("retry_on", None)
33
34 is_valid = False
35 if retry_on in {
36 "5xx",
37 "gateway-error",
38 "connect-failure",
39 "retriable-4xx",
40 "refused-stream",
41 "retriable-status-codes",
42 }:
43 is_valid = True
44
45 return is_valid
46
47 def as_dict(self) -> dict:
48 raw_dict = super().as_dict()
49
50 for key in list(raw_dict):
51 if key in [
52 "_active",
53 "_errored",
54 "_referenced_by",
55 "_rkey",
56 "kind",
57 "location",
58 "name",
59 "namespace",
60 "metadata_labels",
61 ]:
62 raw_dict.pop(key, None)
63
64 return raw_dict
View as plain text