"""The Org aggregate: one org tree (hr_org) with its org units and positions, each sliced over its lifetime.
Invariants, enforced on every change (check()):
* the unit tree is acyclic on every date;
* a unit has at most one chief position at a time, and a second chief for one unit on one date is refused;
* a chief position belongs to the unit it leads, on that date;
* every unit and every position is TC1 over its own lifetime (no gap, no overlap);
* a position moving units is a new slice, not a new key (org_id|position_id|valid_from).
The manager is never stored: ManagerOf (domain/services.py) walks this tree on a date.
"""
from dataclasses import dataclass, field, replace
from datetime import date
from typing import Optional
from brick_hr.domain.errors import Incomplete, RuleBroken
from brick_hr.domain.values import DAY, OPEN
UNIT_FIELDS = ("name", "parent_org_unit_id", "org_unit_type", "cost_centre", "chief_position_id", "location_code")
POSITION_FIELDS = ("org_unit_id", "job_code", "fte_pct", "cost_centre", "vacant_policy")
@dataclass
class UnitSlice:
org_unit_id: str
valid_from: date
valid_to: date = OPEN
name: str = ""
parent_org_unit_id: Optional[str] = None
org_unit_type: Optional[str] = None
cost_centre: Optional[str] = None
chief_position_id: Optional[str] = None
location_code: Optional[str] = None
def covers(self, d: date) -> bool:
return self.valid_from <= d <= self.valid_to
@dataclass
class PositionSlice:
position_id: str
valid_from: date
valid_to: date = OPEN
org_unit_id: str = ""
job_code: Optional[str] = None
fte_pct: Optional[str] = None
cost_centre: Optional[str] = None
vacant_policy: Optional[str] = None
def covers(self, d: date) -> bool:
return self.valid_from <= d <= self.valid_to
@dataclass
class Org:
org_id: str
name: str
legal_entity_ref: str
version: int = 0
units: list = field(default_factory=list)
positions: list = field(default_factory=list)
# ── reading ─────────────────────────────────────────────────────────────────────────────
def unit_at(self, unit_id: Optional[str], d: date) -> Optional[UnitSlice]:
return next((u for u in self.units if u.org_unit_id == unit_id and u.covers(d)), None)
def position_at(self, position_id: Optional[str], d: date) -> Optional[PositionSlice]:
return next((p for p in self.positions if p.position_id == position_id and p.covers(d)), None)
def units_on(self, d: date) -> list:
return sorted((u for u in self.units if u.covers(d)), key=lambda u: u.org_unit_id)
def unit_ids(self) -> set:
return {u.org_unit_id for u in self.units}
def ancestors(self, unit_id: str, d: date) -> list:
"""The unit and every unit above it on `d`, nearest first."""
out, seen, cur = [], set(), unit_id
while cur and cur not in seen:
u = self.unit_at(cur, d)
if u is None:
break
seen.add(cur)
out.append(u)
cur = u.parent_org_unit_id
return out
def descendants(self, unit_id: str, d: date) -> list:
"""Every unit below `unit_id` on `d` (not itself)."""
units = self.units_on(d)
out, frontier = [], [unit_id]
while frontier:
nxt = [u.org_unit_id for u in units if u.parent_org_unit_id in frontier]
nxt = [x for x in nxt if x not in out and x != unit_id]
out += nxt
frontier = nxt
return out
def chief_of(self, unit_id: str, d: date) -> Optional[str]:
u = self.unit_at(unit_id, d)
return u.chief_position_id if u else None
def units_led_by(self, position_ids: set, d: date) -> list:
return [u.org_unit_id for u in self.units_on(d) if u.chief_position_id in position_ids]
# ── writing ─────────────────────────────────────────────────────────────────────────────
@staticmethod
def _slice_change(slices: list, key_attr: str, key: str, valid_from: date, values: dict, make, fields: tuple,
valid_to: Optional[date] = None):
mine = sorted((s for s in slices if getattr(s, key_attr) == key), key=lambda s: s.valid_from)
if any(s.valid_from == valid_from for s in mine):
same = next(s for s in mine if s.valid_from == valid_from)
for f in fields: # a second change on the same day replaces that day's values
if f in values:
setattr(same, f, values[f])
if valid_to is not None:
same.valid_to = valid_to
return same
cur = next((s for s in mine if s.covers(valid_from)), None)
if cur is not None:
new = replace(cur, valid_from=valid_from)
cur.valid_to = valid_from - DAY
else:
later = [s for s in mine if s.valid_from > valid_from]
new = make(key, valid_from)
new.valid_to = (later[0].valid_from - DAY) if later else OPEN
for f in fields:
if f in values:
setattr(new, f, values[f])
if valid_to is not None:
new.valid_to = valid_to
slices.append(new)
return new
def keep_unit(self, unit_id: str, valid_from: date, values: dict, valid_to: Optional[date] = None) -> UnitSlice:
unknown = sorted(set(values) - set(UNIT_FIELDS))
if unknown:
raise Incomplete(f"an org unit has {', '.join(UNIT_FIELDS)}; not {', '.join(unknown)}")
if not unit_id:
raise Incomplete("an org unit needs an id")
before = self.unit_at(unit_id, valid_from)
if "chief_position_id" in values and values["chief_position_id"] and before is not None \
and before.chief_position_id and before.chief_position_id != values["chief_position_id"]:
raise RuleBroken(f"{unit_id} already has a chief position ({before.chief_position_id}) on {valid_from}; "
"a unit has one chief at a time — end that one first (chief_position_id: null)")
s = self._slice_change(self.units, "org_unit_id", unit_id, valid_from, values,
lambda k, f: UnitSlice(k, f), UNIT_FIELDS, valid_to)
if not s.name:
raise Incomplete(f"org unit {unit_id} needs a name")
self.check()
return s
def keep_position(self, position_id: str, valid_from: date, values: dict,
valid_to: Optional[date] = None) -> PositionSlice:
unknown = sorted(set(values) - set(POSITION_FIELDS))
if unknown:
raise Incomplete(f"a position has {', '.join(POSITION_FIELDS)}; not {', '.join(unknown)}")
s = self._slice_change(self.positions, "position_id", position_id, valid_from, values,
lambda k, f: PositionSlice(k, f), POSITION_FIELDS, valid_to)
if not s.org_unit_id or self.unit_at(s.org_unit_id, s.valid_from) is None:
raise RuleBroken(f"position {position_id} must sit in an org unit that exists on {s.valid_from}")
self.check()
return s
# ── invariants ──────────────────────────────────────────────────────────────────────────
def _boundaries(self) -> list:
ds = {u.valid_from for u in self.units} | {p.valid_from for p in self.positions}
ds |= {u.valid_to + DAY for u in self.units if u.valid_to != OPEN}
return sorted(ds)
def check(self) -> None:
for name, slices, attr in (("org unit", self.units, "org_unit_id"), ("position", self.positions, "position_id")):
by: dict = {}
for s in slices:
by.setdefault(getattr(s, attr), []).append(s)
for key, ss in by.items():
ss = sorted(ss, key=lambda s: s.valid_from)
for a, b in zip(ss, ss[1:]):
if b.valid_from <= a.valid_to:
raise RuleBroken(f"{name} {key}: two slices overlap on {b.valid_from}")
if b.valid_from != a.valid_to + DAY:
raise RuleBroken(f"{name} {key}: a gap from {a.valid_to + DAY} to {b.valid_from - DAY}")
for d in self._boundaries():
units = {u.org_unit_id: u for u in self.units_on(d)}
for uid in units: # acyclic on every date
seen, cur = set(), uid
while cur in units:
if cur in seen:
raise RuleBroken(f"moving {uid} there would make a cycle in the org tree on {d} "
f"({' → '.join(sorted(seen))})")
seen.add(cur)
cur = units[cur].parent_org_unit_id
chiefs: dict = {}
for u in units.values():
if not u.chief_position_id:
continue
if u.chief_position_id in chiefs:
raise RuleBroken(f"position {u.chief_position_id} cannot lead both {chiefs[u.chief_position_id]} "
f"and {u.org_unit_id} on {d}")
chiefs[u.chief_position_id] = u.org_unit_id
p = self.position_at(u.chief_position_id, d)
if p is not None and p.org_unit_id != u.org_unit_id:
raise RuleBroken(f"the chief position of {u.org_unit_id} ({u.chief_position_id}) sits in "
f"{p.org_unit_id} on {d}; a chief position belongs to the unit it leads")
__all__ = ["Org", "UnitSlice", "PositionSlice", "UNIT_FIELDS", "POSITION_FIELDS"]