-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmodels.py
More file actions
40 lines (34 loc) · 1.5 KB
/
models.py
File metadata and controls
40 lines (34 loc) · 1.5 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
from pydantic import BaseModel, model_validator
from datetime import datetime
class DelayedRequest(BaseModel):
"""
Represents a delayed HTTP request to be scheduled.
Supports two mutually exclusive scheduling modes:
- delay_seconds: Relative delay from request time (legacy, for logging)
- delay_timestamp: Absolute ISO 8601 timestamp when to execute
For backwards compatibility with existing Redis data:
- Old format only has delay_seconds (required)
- New format can have either field
"""
request_id: str
target_url: str
method: str = 'POST'
headers: dict[str, str]
body: dict | None = None
timestamp: datetime
# Mutually exclusive scheduling fields
delay_seconds: int | None = None
delay_timestamp: datetime | None = None
@model_validator(mode='after')
def check_scheduling_fields(self) -> 'DelayedRequest':
if self.delay_seconds is not None and self.delay_timestamp is not None:
raise ValueError("Only one of 'delay_seconds' or 'delay_timestamp' may be set, not both.")
if self.delay_seconds is None and self.delay_timestamp is None:
raise ValueError("Exactly one of 'delay_seconds' or 'delay_timestamp' must be set.")
return self
def get_display_delay(self) -> str:
"""Get human-readable delay description for logging."""
if self.delay_seconds is not None:
return f"{self.delay_seconds}s"
else:
return f"at {self.delay_timestamp.isoformat()}"