"""hr v4 (2026-10-26, seemo's decided answer 2): what other modules ask hr about where a person belongs — SAP's
organisational assignment plus communication, by person id, so no module types an email or keeps a staff list.
HR-63 GET /persons/{person_id}/assignment?on= company, org unit, team, derived manager and work email on a date
HR-64 GET /roster?company_code=&on=&... who is employed in a company, an org unit or a manager's team
Derived on read from the assignment slice, the ManagerOf walk and the person detail in force on the date; nothing is
stored for the caller, no pay and no sensitive data. The caller is a service client (time-off, expenses,
timesheets, payroll) holding the machine permission; it sees only the people of the companies its grant covers
(View.companies) — anyone else is 404, never 403.
What the spec's words map to here (hr has no separate code for a unit): `team_code` is the org unit the person is
assigned to, `org_unit_code` the unit directly above it on that date (the unit itself when it is a root), and
`org_unit_id` the assigned unit, as everywhere else in hr.
"""
from datetime import date
from typing import Annotated, Optional
from erp_platform.types import BigId
from pydantic import Field
from brick_hr.application import support as S
from brick_hr.application import use_cases as uc
from brick_hr.domain import services
HR = uc.HR
def units_of(view: S.View, org_id: Optional[str], unit_id: Optional[str], d: date) -> tuple:
"""(org_unit_code, team_code) for a person assigned to `unit_id` on `d`."""
if not unit_id:
return None, None
org = view.orgs.get(org_id) if org_id else None
chain = org.ancestors(unit_id, d) if org else []
parent = chain[1].org_unit_id if len(chain) > 1 else unit_id
return parent, unit_id
def assignment_of(view: S.View, person_id: int, d: date, work_email: Optional[str], pool_user_sub: Optional[str]) -> dict:
seat = view.directory.seat(int(person_id), d)
out = {"person_id": int(person_id), "on": d.isoformat(), "employed": seat is not None, "company_code": None,
"org_unit_id": None, "org_unit_code": None, "team_code": None, "manager_person_id": None,
"work_email": None, "pool_user_sub": pool_user_sub or None}
if seat is None:
return out
org_unit_code, team_code = units_of(view, seat.org_id, seat.org_unit_id, d)
out.update(company_code=seat.company_code, org_unit_id=seat.org_unit_id, org_unit_code=org_unit_code,
team_code=team_code, manager_person_id=view.manager_of(int(person_id), d), work_email=work_email or None)
return out
@HR.use_case(kind="query", route="GET /persons/{person_id}/assignment", permission=S.ASSIGNMENT_READ)
def show_assignment(caller: dict, person_id: BigId, on: Optional[S.Day] = None) -> dict:
"""A person's organisational assignment and work email on a date (HR-63, default today in the business
timezone): employed, company_code, org_unit_id, org_unit_code, team_code, the derived manager_person_id,
work_email and pool_user_sub. A date with no engagement answers employed false with nulls, never a 404. A
person outside the companies your grant covers is not found. No pay, no sensitive data."""
with S.rules():
with uc.repository() as uow:
uow.prefetch(person=person_id, seats=True)
view = S.View(uow, caller)
d = on or view.today
p = uow.persons.get(int(person_id))
if p is None or not view.person_in_companies(p.person_id, d):
raise S.not_found(f"person {person_id}")
detail = p.detail.at(d)
return assignment_of(view, p.person_id, d, (detail.data if detail else {}).get("work_email"),
p.pool_user_sub)
PageSize = Annotated[int, Field(ge=1, le=200)] # at most 200 a page (HR-64)
Offset = Annotated[int, Field(ge=0, le=10_000_000)]
@HR.use_case(kind="query", route="GET /roster", permission=S.ROSTER_READ)
def show_roster(caller: dict, company_code: S.Code, on: Optional[S.Day] = None, org_unit_code: Optional[S.Code] = None,
manager_person_id: Optional[BigId] = None, limit: PageSize = 200, offset: Offset = 0) -> dict:
"""Who is employed by a company on a date (HR-64, default today in the business timezone), ordered by
person_id, at most 200 a page with next_offset (null on the last page). org_unit_code narrows to that unit and
every unit below it; manager_person_id to the people whose derived manager they are on that date (direct
reports). People who left before `on` and starters after it are not listed. A company outside the companies
your grant covers answers an empty page, never another company's people. No pay, no sensitive data."""
with S.rules():
with uc.repository() as uow:
uow.prefetch(seats=True)
view = S.View(uow, caller)
d = on or view.today
seats = []
if view.in_companies(company_code):
seats = sorted((s for s in view.directory.seats_on(d) if s.company_code == company_code),
key=lambda s: s.person_id)
if org_unit_code:
seats = [s for s in seats if s.org_unit_id == org_unit_code or org_unit_code in {
u.org_unit_id for u in (view.orgs[s.org_id].ancestors(s.org_unit_id, d)
if s.org_id in view.orgs else [])}]
managers = {s.person_id: view.manager_of(s.person_id, d) for s in seats}
if manager_person_id is not None:
seats = [s for s in seats if managers[s.person_id] == int(manager_person_id)]
page = seats[offset:offset + limit]
emails = uow.persons.names([s.person_id for s in page], d)
people = []
for s in page:
org_unit, team = units_of(view, s.org_id, s.org_unit_id, d)
people.append({"person_id": s.person_id, "company_code": s.company_code, "org_unit_code": org_unit,
"team_code": team, "manager_person_id": managers[s.person_id],
"work_email": (emails.get(s.person_id) or {}).get("work_email") or None,
"fte": str(services.fte_of(s, view.config)),
"employed_from": s.employed_from.isoformat() if s.employed_from else None,
"employed_to": s.employed_to.isoformat() if s.employed_to else None})
more = offset + limit < len(seats)
return {"company_code": company_code, "on": d.isoformat(), "people": people,
"next_offset": offset + limit if more else None}
__all__ = ["show_assignment", "show_roster", "assignment_of", "units_of"]