"""The slice rules (SAP's time constraints), once, for every sliced record.
A TIMELINE is the slices of one series — a person's names, an engagement's assignment, its pay, one pay
component, one kind of sensitive data. The rules, which are in code and never in rows:
* a CHANGE from date F delimits the slice covering F (its valid_to becomes F − 1) and starts a new slice at F
that runs to where the old one ran. Nothing is overwritten: the slice up to F keeps what it said.
* a CORRECTION of the slice starting F is a new slice with the SAME valid_from and a higher `rev`; the old one
is kept with superseded_by = the new rev and superseded_at = when. "What did we believe about date X on
day T" is therefore answerable (believed()).
* a slice that is withdrawn whole (a leave before it starts, a start date moved) is superseded by 0.
* TC1: the active slices of a series cover their span with no gap and no overlap. check_tc1() is called after
every change by the aggregate that owns the series, and refuses in words.
"Current" is never stored: it is the active slice with valid_from ≤ day ≤ valid_to.
"""
from dataclasses import dataclass, field, replace
from datetime import date, datetime
from typing import Iterable, Optional
from brick_hr.domain.errors import RuleBroken
from brick_hr.domain.values import DAY, OPEN
WITHDRAWN = 0
@dataclass
class Slice:
valid_from: date
valid_to: date
data: dict
rev: int = 1
superseded_by: Optional[int] = None
recorded_at: Optional[datetime] = None
superseded_at: Optional[datetime] = None
recorded_by: str = ""
reason: str = ""
@property
def active(self) -> bool:
return self.superseded_by is None
def covers(self, d: date) -> bool:
return self.valid_from <= d <= self.valid_to
def _naive(t: Optional[datetime]) -> Optional[datetime]:
if t is None:
return None
return t.replace(tzinfo=None) if t.tzinfo else t
class Timeline:
"""The slices of one series, active and superseded."""
def __init__(self, name: str, slices: Iterable[Slice] = ()):
self.name = name
self.slices: list = list(slices)
def __repr__(self) -> str:
return f"<Timeline {self.name} {[(s.valid_from, s.valid_to, s.rev, s.superseded_by) for s in self.slices]}>"
# ── reading ─────────────────────────────────────────────────────────────────────────────
def active(self) -> list:
return sorted((s for s in self.slices if s.active), key=lambda s: s.valid_from)
def at(self, d: date) -> Optional[Slice]:
return next((s for s in self.active() if s.covers(d)), None)
def first(self) -> Optional[Slice]:
a = self.active()
return a[0] if a else None
def last(self) -> Optional[Slice]:
a = self.active()
return a[-1] if a else None
def overlapping(self, start: date, end: date) -> list:
return [s for s in self.active() if s.valid_from <= end and start <= s.valid_to]
def believed(self, d: date, known_at: datetime) -> Optional[Slice]:
"""The slice we believed covered `d` at the moment `known_at` (bitemporal read)."""
t = _naive(known_at)
for s in sorted(self.slices, key=lambda s: (s.valid_from, s.rev)):
rec, sup = _naive(s.recorded_at), _naive(s.superseded_at)
if rec is not None and rec > t:
continue
if sup is not None and sup <= t:
continue
if s.covers(d):
return s
return None
def _next_rev(self, valid_from: date) -> int:
return max([s.rev for s in self.slices if s.valid_from == valid_from] or [0]) + 1
# ── writing ─────────────────────────────────────────────────────────────────────────────
def start(self, valid_from: date, data: dict, *, now: datetime, by: str, reason: str = "",
valid_to: date = OPEN) -> Slice:
"""The first slice of a series (or one after a gap the owner allows, e.g. a new engagement)."""
if any(s.valid_from <= valid_to and valid_from <= s.valid_to for s in self.active()):
raise RuleBroken(f"{self.name}: a slice already covers part of {valid_from} – {valid_to}")
s = Slice(valid_from, valid_to, dict(data), self._next_rev(valid_from), recorded_at=now, recorded_by=by,
reason=reason)
self.slices.append(s)
return s
def change(self, valid_from: date, data: dict, *, now: datetime, by: str, reason: str = "",
merge: bool = True, until: Optional[date] = None) -> Slice:
"""A change from `valid_from`: delimit the covering slice, start a new one where it ran to."""
if any(s.valid_from == valid_from for s in self.active()):
raise RuleBroken(f"{self.name}: a slice already starts on {valid_from} — correct it instead of "
"recording a second change on the same day")
cur = self.at(valid_from)
if cur is None:
last = self.last()
if last is not None and valid_from > last.valid_to + DAY:
# found by the TC1 property test: a change after the series ended would open a gap
raise RuleBroken(f"{self.name}: ends {last.valid_to}; a change on {valid_from} would leave a gap")
later = [s for s in self.active() if s.valid_from > valid_from]
to = (later[0].valid_from - DAY) if later else OPEN
base = {}
else:
to = cur.valid_to
base = dict(cur.data)
cur.valid_to = valid_from - DAY
if until is not None:
to = min(to, until)
new = Slice(valid_from, to, ({**base, **data} if merge else dict(data)), self._next_rev(valid_from),
recorded_at=now, recorded_by=by, reason=reason)
self.slices.append(new)
return new
def correct(self, valid_from: date, data: dict, *, now: datetime, by: str, reason: str = "") -> Slice:
"""Supersede the active slice starting `valid_from` with a corrected one; the old one stays readable."""
old = next((s for s in self.active() if s.valid_from == valid_from), None)
if old is None:
raise RuleBroken(f"{self.name}: no slice starts on {valid_from} to correct")
new = Slice(old.valid_from, old.valid_to, {**old.data, **data}, self._next_rev(valid_from),
recorded_at=now, recorded_by=by, reason=reason or "correction")
old.superseded_by, old.superseded_at = new.rev, now
self.slices.append(new)
return new
def end_at(self, last_day: date, *, now: datetime) -> list:
"""Nothing of this series after `last_day`: the covering slice ends that day (delimited at last_day + 1),
a slice starting later is withdrawn. Returns the slices touched."""
touched = []
for s in self.active():
if s.valid_from > last_day:
s.superseded_by, s.superseded_at = WITHDRAWN, now
touched.append(s)
elif s.valid_to > last_day:
s.valid_to = last_day
touched.append(s)
return touched
def reopen(self, old_end: date, new_end: date) -> Optional[Slice]:
"""The last slice ran to `old_end` (an engagement's last day); it now runs to `new_end`."""
last = self.last()
if last is not None and last.valid_to == old_end:
last.valid_to = new_end
return last
return None
def move_start(self, old_start: date, new_start: date, *, now: datetime, by: str) -> Optional[Slice]:
"""The series starts on `new_start` instead of `old_start` (a start date moved): the first slice is
withdrawn and re-recorded at the new date, so the original stays readable."""
first = self.first()
if first is None or first.valid_from != old_start or old_start == new_start:
return None
if new_start > first.valid_to:
raise RuleBroken(f"{self.name}: the first slice ends {first.valid_to}; a start on {new_start} would "
"skip it — record the later change again after moving the date")
moved = Slice(new_start, first.valid_to, dict(first.data), self._next_rev(new_start), recorded_at=now,
recorded_by=by, reason=f"start moved from {old_start}")
first.superseded_by, first.superseded_at = WITHDRAWN, now
self.slices.append(moved)
return moved
# ── TC1 ─────────────────────────────────────────────────────────────────────────────────
def check_tc1(self, start: Optional[date] = None, end: Optional[date] = None) -> None:
"""No gap and no overlap among the active slices; with start/end, they cover exactly that span."""
a = self.active()
if not a:
if start is not None:
raise RuleBroken(f"{self.name}: nothing covers {start} onwards")
return
for prev, nxt in zip(a, a[1:]):
if nxt.valid_from <= prev.valid_to:
raise RuleBroken(f"{self.name}: slices overlap on {nxt.valid_from} "
f"({prev.valid_from}–{prev.valid_to} and {nxt.valid_from}–{nxt.valid_to})")
if nxt.valid_from != prev.valid_to + DAY:
raise RuleBroken(f"{self.name}: a gap from {prev.valid_to + DAY} to {nxt.valid_from - DAY}")
if start is not None and a[0].valid_from != start:
raise RuleBroken(f"{self.name}: must start on {start}, starts on {a[0].valid_from}")
if end is not None and a[-1].valid_to != end:
raise RuleBroken(f"{self.name}: must run to {end}, runs to {a[-1].valid_to}")
def clone(t: Timeline) -> Timeline:
return Timeline(t.name, [replace(s, data=dict(s.data)) for s in t.slices])
__all__ = ["Slice", "Timeline", "WITHDRAWN", "clone", "field"]