blob: 564a32df6796632e611507382f0e6780d8178e61 (
plain)
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
41
42
43
44
45
46
47
48
|
from typing import Optional, Dict
from pydantic import BaseModel, Field, ConfigDict, PositiveInt
from typing_extensions import Self
from jb.models.currency import USDCent
from jb.models.custom_types import AMTBoto3ID, AwareDatetimeISO, UUIDStr
from jb.models.definitions import PayoutStatus
class Bonus(BaseModel):
"""
A Bonus is created (in our DB) ONLY associated with an APPROVED
thl-payout-event, AFTER the bonus has actually been sent to
the worker.
We have the payout_event uuid as the unique request token to make
sure it only gets sent once (param in the boto request).
"""
model_config = ConfigDict(
extra="forbid",
validate_assignment=True,
)
id: Optional[PositiveInt] = Field(default=None)
assignment_id: Optional[PositiveInt] = Field(default=None)
amt_worker_id: str = Field(min_length=3, max_length=50)
amt_assignment_id: AMTBoto3ID = Field()
amount: USDCent = Field()
reason: str = Field(min_length=5)
grant_time: AwareDatetimeISO = Field()
# -- GRL Specific ---
payout_event_id: UUIDStr = Field()
# created: Optional[AwareDatetimeISO] = Field(default=None)
def to_postgres(self):
d = self.model_dump(mode="json")
d["amount"] = self.amount.to_usd()
return d
@classmethod
def from_postgres(cls, data: Dict) -> Self:
data["amount"] = USDCent(round(data["amount"] * 100))
fields = set(cls.model_fields.keys())
data = {k: v for k, v in data.items() if k in fields}
return cls.model_validate(data)
|
