aaronrene / scooling-lab public
retention.py python
131 lines 4.2 KB
Raw
sha256:ca1d0e687bff8686b37126eb8e4fb6d38b40e189352a4920d66ae89c7274e340 Add T4 job cancellation retry and validation Human minor ⚠ breaking 41 days ago
1 """Retention policy classes and expiry evaluation for fixture artifacts."""
2
3 from __future__ import annotations
4
5 from dataclasses import dataclass
6 from datetime import UTC, datetime, timedelta
7 from enum import Enum
8 from types import MappingProxyType
9 from typing import Mapping
10
11 from scooling_lab.errors import ApiError, ErrorCode
12
13
14 class RetentionPolicyClass(str, Enum):
15 """Supported artifact retention classes for synthetic fixture jobs."""
16
17 EPHEMERAL = "ephemeral"
18 STANDARD = "standard"
19 EXTENDED = "extended"
20
21
22 @dataclass(frozen=True)
23 class RetentionBounds:
24 """Default and allowed TTL range for one retention class."""
25
26 default_ttl_seconds: int
27 min_ttl_seconds: int
28 max_ttl_seconds: int
29
30
31 @dataclass(frozen=True)
32 class RetentionPolicy:
33 """Validated retention policy for one training job request."""
34
35 policy_class: RetentionPolicyClass
36 ttl_seconds: int
37
38 def to_public_dict(self) -> dict[str, str | int]:
39 """Serialize the policy without private paths, text, or credentials."""
40
41 return {
42 "policyClass": self.policy_class.value,
43 "ttlSeconds": self.ttl_seconds,
44 }
45
46
47 RETENTION_BOUNDS: MappingProxyType[RetentionPolicyClass, RetentionBounds] = MappingProxyType(
48 {
49 RetentionPolicyClass.EPHEMERAL: RetentionBounds(
50 default_ttl_seconds=3_600,
51 min_ttl_seconds=60,
52 max_ttl_seconds=86_400,
53 ),
54 RetentionPolicyClass.STANDARD: RetentionBounds(
55 default_ttl_seconds=2_592_000,
56 min_ttl_seconds=3_600,
57 max_ttl_seconds=7_776_000,
58 ),
59 RetentionPolicyClass.EXTENDED: RetentionBounds(
60 default_ttl_seconds=31_536_000,
61 min_ttl_seconds=86_400,
62 max_ttl_seconds=63_072_000,
63 ),
64 }
65 )
66
67
68 def default_retention_policy() -> RetentionPolicy:
69 """Return the default artifact retention policy for fixture jobs."""
70
71 bounds = RETENTION_BOUNDS[RetentionPolicyClass.STANDARD]
72 return RetentionPolicy(
73 policy_class=RetentionPolicyClass.STANDARD,
74 ttl_seconds=bounds.default_ttl_seconds,
75 )
76
77
78 def retention_policy_from_mapping(value: object) -> RetentionPolicy:
79 """Validate a request retention policy or return the default policy."""
80
81 if value is None:
82 return default_retention_policy()
83 if not isinstance(value, Mapping):
84 raise ApiError(ErrorCode.VALIDATION_ERROR, 400)
85 if set(value).difference({"policyClass", "ttlSeconds"}):
86 raise ApiError(ErrorCode.VALIDATION_ERROR, 400)
87
88 policy_class_value = value.get("policyClass")
89 if not isinstance(policy_class_value, str):
90 raise ApiError(ErrorCode.VALIDATION_ERROR, 400)
91 try:
92 policy_class = RetentionPolicyClass(policy_class_value)
93 except ValueError as exc:
94 raise ApiError(ErrorCode.VALIDATION_ERROR, 400) from exc
95
96 bounds = RETENTION_BOUNDS[policy_class]
97 ttl_value = value.get("ttlSeconds", bounds.default_ttl_seconds)
98 if not isinstance(ttl_value, int) or isinstance(ttl_value, bool):
99 raise ApiError(ErrorCode.VALIDATION_ERROR, 400)
100 if not bounds.min_ttl_seconds <= ttl_value <= bounds.max_ttl_seconds:
101 raise ApiError(ErrorCode.VALIDATION_ERROR, 400)
102
103 return RetentionPolicy(policy_class=policy_class, ttl_seconds=ttl_value)
104
105
106 def parse_utc_timestamp(value: str) -> datetime:
107 """Parse a public UTC timestamp emitted by Scooling Lab."""
108
109 if not value.endswith("Z"):
110 raise ApiError(ErrorCode.INTERNAL_ERROR, 500)
111 return datetime.fromisoformat(value.removesuffix("Z") + "+00:00").astimezone(UTC)
112
113
114 def utc_timestamp(value: datetime) -> str:
115 """Format a UTC datetime for public API metadata."""
116
117 return value.astimezone(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z")
118
119
120 def expires_at(created_at: str, policy: RetentionPolicy) -> str:
121 """Return the expiry timestamp for an artifact under a retention policy."""
122
123 expires = parse_utc_timestamp(created_at) + timedelta(seconds=policy.ttl_seconds)
124 return utc_timestamp(expires)
125
126
127 def is_expired(created_at: str, policy: RetentionPolicy, now: datetime) -> bool:
128 """Return whether an artifact is expired at the supplied UTC instant."""
129
130 expiry = parse_utc_timestamp(expires_at(created_at, policy))
131 return now.astimezone(UTC) >= expiry
File History 1 commit
sha256:ca1d0e687bff8686b37126eb8e4fb6d38b40e189352a4920d66ae89c7274e340 Add T4 job cancellation retry and validation Human minor ⚠ 41 days ago