"""Domain events: the seven kinds of personnel event, past tense. They ARE the hr_event rows.
Hired a new engagement starts (first ever, for a new person). Starts an onboarding run.
Transferred a new assignment slice. Starts a change run (the access delta).
PayChanged a new pay or component slice; reason required. No run (the payroll extract carries it).
ContractExtended planned_end moves. Re-times an open offboarding run's undone steps.
Left actual_end set; assignment and pay delimited at last day + 1. Starts an offboarding run,
unless the reason is 'converted'.
Rehired a NEW engagement n+1 for an existing person; history kept. Starts an onboarding run (the
rehire playbook). NEVER starts an offboarding run.
Converted Left (reason converted) on n plus a new engagement n+1 on the same date, as ONE event.
Starts one conversion run; access is kept.
What each kind writes and starts is code, not rows (in_code_not_rows). Events are append-only; each writes at
least one slice in the same transaction.
"""
from dataclasses import dataclass, field
from datetime import date, datetime
from typing import Optional
from brick_hr.domain.values import DAY
HIRED, TRANSFERRED, PAY_CHANGED, CONTRACT_EXTENDED, LEFT, REHIRED, CONVERTED = (
"Hired", "Transferred", "PayChanged", "ContractExtended", "Left", "Rehired", "Converted")
KINDS = (HIRED, TRANSFERRED, PAY_CHANGED, CONTRACT_EXTENDED, LEFT, REHIRED, CONVERTED)
CONVERTED_REASON = "converted"
def direction_of(kind: str, reason: Optional[str]) -> Optional[str]:
"""Which run a committed event starts, if any. A rehire never offboards; a conversion keeps access."""
if kind in (HIRED, REHIRED):
return "onboarding"
if kind == CONVERTED:
return "conversion"
if kind == TRANSFERRED:
return "change"
if kind == LEFT:
return None if (reason or "") == CONVERTED_REASON else "offboarding"
return None # PayChanged, ContractExtended
@dataclass
class Event:
n: int
seq: int
event_type: str
kind: str
reason: Optional[str]
effective_date: date
actor: str
recorded_at: datetime
slices_written: list = field(default_factory=list)
run_id: Optional[str] = None
bp_synced_at: Optional[datetime] = None
payload: dict = field(default_factory=dict)
@property
def takes_effect_on(self) -> date:
"""When the outside world (the business partner) must see it: a leaver loses the role AFTER the last day."""
return self.effective_date + DAY if self.kind == LEFT else self.effective_date
@property
def direction(self) -> Optional[str]:
return direction_of(self.kind, self.reason)
__all__ = ["Event", "KINDS", "HIRED", "TRANSFERRED", "PAY_CHANGED", "CONTRACT_EXTENDED", "LEFT", "REHIRED",
"CONVERTED", "CONVERTED_REASON", "direction_of"]