"""Value objects: immutable, compared by value, each carrying the rule that makes it valid.
EffectivePeriod valid_from / valid_to (9999-12-31 = open); delimit, covers, overlaps
PersonKey 'staff:<no>' else 'email:<lower(trim(personal email))>' — fixed at creation, never re-keyed
PersonId int.from_bytes(blake2b(person_key, 7)): a 56-bit BIGINT, like bp_id
EngagementRef person_id|n
RunId <direction>:<person_id>:<n>:<event seq>
Minutes an integer count of minutes (time is never a float)
Days NUMERIC(5,2)
Rate erp_platform.money.Rate: NUMERIC(19,4) + basis + currency, never converted at write time
"""
import hashlib
import re
from dataclasses import dataclass
from datetime import date, timedelta
from decimal import Decimal, InvalidOperation
from typing import Optional
from erp_platform.money import BASES, InvalidMoney, Rate
from brick_hr.domain.errors import Incomplete
OPEN = date(9999, 12, 31)
DAY = timedelta(days=1)
def as_date(value, what: str = "date") -> date:
"""A date from a date or an ISO string; a datetime is not a date."""
if isinstance(value, date) and not hasattr(value, "hour"):
return value
if isinstance(value, str):
try:
return date.fromisoformat(value.strip()[:10])
except ValueError:
pass
raise Incomplete(f"{what} is a date like 2026-11-02, not {value!r}")
@dataclass(frozen=True)
class EffectivePeriod:
valid_from: date
valid_to: date = OPEN
def __post_init__(self) -> None:
if self.valid_to < self.valid_from:
raise Incomplete(f"a period cannot end ({self.valid_to}) before it starts ({self.valid_from})")
@property
def is_open(self) -> bool:
return self.valid_to == OPEN
def covers(self, d: date) -> bool:
return self.valid_from <= d <= self.valid_to
def overlaps(self, other: "EffectivePeriod") -> bool:
return self.valid_from <= other.valid_to and other.valid_from <= self.valid_to
def delimit(self, new_from: date) -> "EffectivePeriod":
"""End this period the day before `new_from` (valid_to = new_from − 1). Nothing is overwritten:
the period up to then stays what it was."""
if new_from <= self.valid_from:
raise Incomplete(f"cannot delimit a period starting {self.valid_from} at {new_from}")
return EffectivePeriod(self.valid_from, new_from - DAY)
def clamp(self, start: date, end: date) -> Optional["EffectivePeriod"]:
lo, hi = max(self.valid_from, start), min(self.valid_to, end)
return EffectivePeriod(lo, hi) if lo <= hi else None
_EMAIL = re.compile(r"[^@\s]+@[^@\s]+\.[^@\s]+")
def normalise_email(email: str, what: str = "email") -> str:
e = (email or "").strip().lower()
if not _EMAIL.fullmatch(e):
raise Incomplete(f"{what} is an address like anna@example.org, not {email!r}")
return e
@dataclass(frozen=True)
class PersonKey:
"""Who a person is, in the customer's terms: the staff number, else the normalised personal email.
Fixed when the person is created: a staff number given later is stored, not re-keyed."""
value: str
@classmethod
def of(cls, staff_no: Optional[str], personal_email: Optional[str]) -> "PersonKey":
no = (staff_no or "").strip()
if no:
return cls(f"staff:{no.lower()}")
if not (personal_email or "").strip():
raise Incomplete("a person needs a staff number or a personal email to be identified by")
return cls(f"email:{normalise_email(personal_email, 'personal_email')}")
@property
def person_id(self) -> int:
return PersonId.of(self)
class PersonId:
@staticmethod
def of(key: PersonKey) -> int:
return int.from_bytes(hashlib.blake2b(key.value.encode(), digest_size=7).digest(), "big")
def engagement_ref(person_id: int, n: int) -> str:
return f"{int(person_id)}|{int(n)}"
DIRECTIONS = ("onboarding", "offboarding", "change", "conversion", "invite", "warning")
def run_id(direction: str, person_id: int, n: int, seq) -> str:
"""Derived, so a double click (or a retried tick) is one run."""
if direction not in DIRECTIONS:
raise ValueError(f"direction is one of {DIRECTIONS}")
return f"{direction}:{int(person_id)}:{int(n)}:{seq}"
def parse_run_id(rid: str) -> tuple:
parts = (rid or "").split(":")
if len(parts) != 4 or parts[0] not in DIRECTIONS:
raise Incomplete(f"{rid!r} is not a run id (<direction>:<person_id>:<n>:<seq>)")
try:
return parts[0], int(parts[1]), int(parts[2]), parts[3]
except ValueError:
raise Incomplete(f"{rid!r} is not a run id (<direction>:<person_id>:<n>:<seq>)") from None
@dataclass(frozen=True)
class Minutes:
value: int
def __post_init__(self) -> None:
if isinstance(self.value, bool) or not isinstance(self.value, int) or self.value < 0:
raise Incomplete(f"minutes are a whole number ≥ 0 (time is never a float), not {self.value!r}")
def days(value) -> Decimal:
"""Days as NUMERIC(5,2): two places at most, never a float."""
if isinstance(value, float):
raise Incomplete("days are an exact number (\"0.5\"), never a float")
try:
d = Decimal(str(value))
except InvalidOperation:
raise Incomplete(f"days is a number, not {value!r}") from None
if d != d.quantize(Decimal("0.01")) or abs(d) >= 1000:
raise Incomplete(f"days carry two places and fit NUMERIC(5,2), not {value!r}")
return d.quantize(Decimal("0.01"))
def rate(amount, basis: str, currency: str) -> Rate:
"""A pay rate: 4 places, a basis from hour/day/month/year/fixed, a currency. Never converted between bases."""
if amount is None or basis is None or currency is None:
raise Incomplete("a rate needs an amount, a basis and a currency")
if isinstance(amount, float):
raise Incomplete("a rate is an exact number (\"45000.0000\"), never a float")
try:
return Rate(Decimal(str(amount)), str(basis).strip().lower(), str(currency))
except (InvalidMoney, InvalidOperation, ValueError) as e:
raise Incomplete(str(e)) from None
__all__ = ["OPEN", "DAY", "BASES", "EffectivePeriod", "PersonKey", "PersonId", "engagement_ref", "run_id",
"parse_run_id", "Minutes", "days", "rate", "as_date", "normalise_email"]