dodil/erp-brick-timesheets
DODIL ERP brick: timesheets
src/brick_timesheets/infrastructure/seams.py
"""The neighbours this brick asks, each behind ONE adapter - the domain never knows how they are reached.
hr the person (by the caller's sub), their employment and work pattern on dates, and their derived manager on a date.
Timesheets keeps no staff list: no person in hr, no timesheet.
projects what time can be booked to on a day, and a project's time terms (billing, how it bills, cost centre, lead).
time-off the approved absences of a week, which become read-only lines.
gl the ledger: an approved week posts ONE labour journal whose id is derived from the sheet, revision and event; cost centres.
sales-order the delivery: approved billable hours on a sales-order project are delivered on the order's service line.
product the price of a time product on a work day (the billing run of a contract or manual project).
ar the draft invoice of a released billing run: ONE invoice per run, replayed by ar on the run's source id.
What they serve (the callee's signed spec, erp-method modules/<id>/module.yaml seams; tests/conftest.py checks every call):
hr GET /persons?pool_user_sub= -> {person_id, bp_id, company_code}; 404 when no person carries the sub (HR-58)
GET /persons/{person_id}/employment?on= -> {employed, company_code, cost_centre, ...} (HR-56)
GET /persons/{person_id}/manager?on=&level= -> {manager: {person_id, pool_user_sub, email} | null} (HR-57)
GET /persons/{person_id}/work-pattern?from=&to= -> {days: [{date, employed, contracted_minutes, public_holiday}]} (HR-62)
projects GET /projects/bookable?person_id=&on= -> {projects: [{project_id, billing, items: [{item_id, state}]}]} (PJ-25)
GET /projects/{project_id}/time-terms?on= -> {state, billing, bill_via, sales_order_id, products, cost_centre, lead_person_id}
time-off GET /absences?company_code=&from=&to=&person_id= -> {absences: [{absence_ref, leave_type_code, status, days: [...]}]} (TOF-20)
gl POST /journal-entry/journals -> 201 {journal_id, replay: false}; 200 replay; 409 {gate: period_open} (GL-30)
GET /cost-centres/{cost_centre} -> {status: open | closed}; 404 when unknown (GL-61)
sales-order POST /sales-orders/{sales_order_id}/deliveries -> the order view; 409 {gate: within_ordered | service_line | not_below_invoiced} (SO-51)
product GET /products/{product_id}/price?quantity=&unit_code=&bp_id=¤cy=&priced_on=
-> {unit_price, pricing_quantity, pricing_unit_code, priced_from, line: {tax_category}} (PRD-31)
ar POST /invoices/raise -> 201 the draft invoice {invoice_id}; 200 the same for the same source_id; 422 {gate: not_negative} (AR-20)
Every answer is a reply, never an exception: not installed (409), failed (502), did not answer (503).
Tests register in-memory fakes with `erp_platform.seam.register_local("hr", app)`.
"""
import json
from datetime import date
from typing import Optional
from urllib.parse import quote
from erp_platform.capabilities import CapabilitySeam, ManualTask
from erp_platform.seam import Seam, derived_id
MAIL = CapabilitySeam("mail.send") # a reminder is an email through the composition's mail provider, or a task for a person when it has none
HR = Seam("hr", "HR_BASE_URL", roles=["service"], permissions=["hr:employment:read", "hr:manager:read", "hr:work_pattern:read"])
PROJECTS = Seam("projects", "PROJECTS_BASE_URL", roles=["service"], permissions=["pj:time_terms:read"])
TIME_OFF = Seam("time-off", "TIME_OFF_BASE_URL", roles=["service"], permissions=["tof:absence:read_for_payroll"])
LEDGER = Seam("gl", "GL_BASE_URL", roles=["service"], permissions=["gl:journal:post", "gl:cost_centre:read"])
ORDERS = Seam("sales-order", "SALES_ORDER_BASE_URL", roles=["service"], permissions=["so:order:deliver"])
PRODUCT = Seam("product", "PRODUCT_BASE_URL", roles=["service"], permissions=["product:price_list:read"])
INVOICES = Seam("ar", "AR_BASE_URL", roles=["service"], permissions=["ar:invoice:raise"])
class Unavailable(Exception):
"""A neighbour could not answer. `http` is what this brick answers: 409 not installed, 502 failed, 503 silent."""
def __init__(self, http: int, reason: str):
super().__init__(reason)
self.http, self.reason = http, reason
def journal_id(source_document: str, event: str) -> str:
"""The journal's id, DERIVED (gl checks it): posting twice is gl's replay, never a second journal."""
return str(derived_id(source_document, event))
def _body(text: str) -> dict:
try:
got = json.loads(text)
return got if isinstance(got, dict) else {}
except (ValueError, TypeError):
return {}
def _read(seam: Seam, name: str, path: str, params: dict, actor: str, why: str) -> Optional[dict]:
"""A 200 body, None for a 404, or Unavailable."""
if not seam.installed:
raise Unavailable(409, f"{name} is not part of this installation: {why}")
r = seam.get(path, params=params, actor=actor)
if r.ok:
return r.data
if r.status == 404:
return None
if r.status and r.status >= 500:
raise Unavailable(502, f"{name} failed ({r.status}: {r.reason or 'no reason given'}) - try again")
if r.status:
raise Unavailable(502, f"{name} answered {r.status} ({r.reason or 'no reason given'})")
raise Unavailable(503, f"{name} did not answer ({r.reason or 'no reason given'}) - try again")
def _hr(path: str, params: dict, actor: str) -> Optional[dict]:
return _read(HR, "hr", path, params, actor, "the person, their hours and their manager come from hr, so nobody can record time")
def person_by_sub(sub: str, actor: str = "") -> Optional[dict]:
"""The signed-in user's person: {person_id, bp_id, company_code}, or None when hr holds nobody with that sub."""
return _hr("/persons", {"pool_user_sub": sub}, actor)
def employment(person_id: int, on: date, actor: str = "") -> dict:
"""The person's employment on a date: {employed, company_code, cost_centre, ...} (employed False when hr has no such person)."""
return _hr(f"/persons/{person_id}/employment", {"on": on.isoformat()}, actor) or {"employed": False}
def manager(person_id: int, on: date, level: int, actor: str = "") -> Optional[dict]:
"""hr's derived manager `level` steps up on a date: {person_id, bp_id, pool_user_sub, email}, or None."""
got = _hr(f"/persons/{person_id}/manager", {"on": on.isoformat(), "level": level}, actor)
return (got or {}).get("manager")
def work_pattern(person_id: int, start: date, end: date, actor: str = "") -> list:
"""hr's days of the range: [{date, employed, contracted_minutes, public_holiday, ...}]; [] when hr has no such person."""
got = _hr(f"/persons/{person_id}/work-pattern", {"from": start.isoformat(), "to": end.isoformat()}, actor)
return (got or {}).get("days") or []
def bookable(person_id: int, on: date, actor: str = "") -> list:
"""The projects this person may book to on a day, each with its open items: [{project_id, name, bp_id, billing, items: [{item_id, title, state}]}]."""
got = _read(PROJECTS, "projects", "/projects/bookable", {"person_id": person_id, "on": on.isoformat()}, actor,
"time can be booked to cost centres only")
return (got or {}).get("projects") or []
def time_terms(project_id: str, on: date, actor: str = "") -> Optional[dict]:
"""One project's time terms on a day, or None for an unknown project."""
return _read(PROJECTS, "projects", f"/projects/{quote(project_id, safe=':@')}/time-terms", {"on": on.isoformat()}, actor,
"time can be booked to cost centres only")
def absences(company_code: str, start: date, end: date, person_id: int, actor: str = "") -> list:
"""The person's approved and recorded absences of the range (time-off). Without time-off there are none: holidays are booked as time."""
if not TIME_OFF.installed:
return []
got = _read(TIME_OFF, "time-off", "/absences", {"company_code": company_code, "from": start.isoformat(), "to": end.isoformat(),
"person_id": person_id}, actor, "holidays are not filled in")
return [a for a in (got or {}).get("absences") or [] if a.get("status") in ("approved", "recorded")]
def cost_centre_open(cost_centre: str, actor: str = "") -> Optional[bool]:
"""Whether gl holds the cost centre open: True / False, None when gl is not installed (nothing to check against) or does not know it."""
if not LEDGER.installed:
return None
got = _read(LEDGER, "gl", f"/cost-centres/{quote(cost_centre, safe='')}", {}, actor, "")
if got is None:
return False
return got.get("status") == "open"
def _outcome(r) -> dict:
"""What a neighbour said to a write: ok, or {gate, reason, fix} for a refusal, or a transient failure (retry)."""
if r.ok:
return {"ok": True, "data": r.data, "retry": False}
if r.status and 400 <= r.status < 500:
body = _body(r.reason)
return {"ok": False, "retry": False, "status": r.status, "gate": body.get("gate", ""), "reason": body.get("reason") or r.reason or f"answered {r.status}",
"fix": body.get("fix", "")}
why = f"failed ({r.status}: {r.reason or 'no reason given'})" if r.status else f"did not answer ({r.reason or 'no reason given'})"
return {"ok": False, "retry": True, "status": r.status or 0, "gate": "", "reason": why, "fix": "it is retried on the next pass"}
def post_journal(body: dict, actor: str) -> dict:
"""POST the stored journal body to gl, unchanged. A 409/422 is a refusal (the period is closed: gl's gate and reason); no answer is a retry."""
if not LEDGER.installed:
return {"ok": False, "retry": True, "status": 0, "gate": "", "reason": "no ledger is installed, so nothing was posted", "fix": "install gl"}
return _outcome(LEDGER.post("/journal-entry/journals", body, actor=actor))
def post_delivery(sales_order_id: str, body: dict, actor: str) -> dict:
"""POST the stored delivery to the order, unchanged. The order replays the same delivery_ref; a refusal (within_ordered ...) is its gate and reason."""
if not ORDERS.installed:
return {"ok": False, "retry": True, "status": 0, "gate": "", "reason": "sales-order is not part of this installation, so the hours were not delivered",
"fix": "install sales-order"}
return _outcome(ORDERS.post(f"/sales-orders/{quote(sales_order_id, safe=':@')}/deliveries", body, actor=actor))
def price(product_id: str, quantity: str, bp_id: int, currency: str, priced_on: date, actor: str = "") -> Optional[dict]:
"""The product's price of `quantity` hours for this customer on a day (TS-14), or None for an unknown product / no price. Never an invented price."""
return _read(PRODUCT, "product", f"/products/{quote(product_id, safe='')}/price",
{"quantity": quantity, "unit_code": "HUR", "bp_id": bp_id, "currency": currency, "priced_on": priced_on.isoformat()}, actor,
"the time product's price comes from product, so nothing can be billed through ar")
def post_invoice(body: dict, actor: str) -> dict:
"""POST the stored invoice request to ar, unchanged. ar replays the same (source_kind, source_id); a refusal (not_negative ...) is its gate and reason."""
if not INVOICES.installed:
return {"ok": False, "retry": True, "status": 0, "gate": "", "reason": "ar is not part of this installation, so the run was not invoiced", "fix": "install ar"}
return _outcome(INVOICES.post("/invoices/raise", body, actor=actor))
def send_mail(key: str, from_email: str, to: list, subject: str, body: str, actor: str) -> dict:
"""Hand one reminder to the mail capability (idempotent on `key`: a retried pass is the stored answer). {ok, manual, retry, reason}: with no provider it
is a task for a person (manual), a provider that is down is retried."""
res = MAIL.call(key, {"from_email": from_email, "to": to, "cc": [], "subject": subject[:200], "body_text": body, "body_html": None}, actor=actor)
if isinstance(res, ManualTask):
return {"ok": True, "manual": True, "retry": False, "reason": res.reason}
if res.status == "done":
return {"ok": True, "manual": False, "retry": False, "reason": ""}
return {"ok": False, "manual": False, "retry": bool(res.retryable), "reason": f"mail capability {res.status}: {res.message or ''}".strip()}