"""Documents on a person (HR-20): typed, with an expiry where the type has one, and an ACCESS CLASS from the type:
personnel (hr:document:read), pay (hr:pay:read — the signed contract) or sensitive (hr:sensitive:read — filed in
the Sensitive aggregate, like the right-to-work copy). The bytes are STREAMED by these use cases after the class
check (base64 in the answer); a direct or presigned object link is never handed out. A document of a class the
caller may not read is not found.
"""
import base64
import hashlib
from typing import Optional
from erp_platform.bricks import Refused
from erp_platform.types import BigId, Name, text
from brick_hr.application import engine
from brick_hr.application import support as S
from brick_hr.application import use_cases as uc
from brick_hr.domain.sensitive import Sensitive
HR = uc.HR
MAX_BYTES = 10 * 1024 * 1024
def _may_read(caller: dict, access_class: str) -> bool:
return {"personnel": S.has(caller, S.DOC_READ) or S.has(caller, S.PAY_READ),
"pay": S.has(caller, S.PAY_READ), "sensitive": S.has(caller, S.SENSITIVE_READ)}.get(access_class, False)
@HR.use_case(kind="command", route="POST /people/{person_id}/documents", permission=S.DOC_WRITE, idempotent=True)
def file_document(caller: dict, person_id: BigId, document_type: Name, filename: Name, content_base64: str,
content_type: text(100) = "application/pdf", expires_on: Optional[S.Day] = None) -> dict:
"""File a document on a person: its type decides the access class (personnel, pay or sensitive) and whether it
expires. A sensitive type (right-to-work evidence) is filed in the Sensitive aggregate and needs
hr:sensitive:write. The same file twice is one document (its id is derived from its sha256)."""
at = S.now()
try:
content = base64.b64decode(content_base64, validate=True)
except ValueError:
raise Refused(422, "content_base64 is not base64") from None
if len(content) > MAX_BYTES:
raise Refused(413, "the document is larger than 10 MiB")
sha = hashlib.sha256(content).hexdigest()
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller, at)
view.person(person_id)
dt = view.config.document_type(document_type)
if dt is None:
raise Refused(422, f"there is no document type {document_type!r}")
if dt.get("expires") and expires_on is None and dt["access_class"] != "sensitive":
raise Refused(422, f"a {dt['label']} has an expiry date")
if dt["access_class"] == "sensitive":
if not S.has(caller, S.SENSITIVE_WRITE):
raise Refused(404, f"no person {person_id}")
key = f"hr/{person_id}/sensitive/{dt['sensitive_kind']}-{sha[:16]}"
s = uow.sensitive.get(person_id) or Sensitive(person_id)
held = (s.current(view.today).get(dt["sensitive_kind"]) or {})
t = s.kinds.get(dt["sensitive_kind"])
if not (t and any(sl.data.get("sha256") == sha for sl in t.active())):
s.record(dt["sensitive_kind"], {"filename": filename, "content_type": content_type,
"size_bytes": len(content),
"expires_on": expires_on.isoformat() if expires_on else None},
view.today, now=at, by=S.actor(caller), object_key=key, sha256=sha)
uow.objects.put(key, content, content_type)
uow.sensitive.save(s)
del held
return {"person_id": person_id, "doc_id": f"sensitive:{dt['sensitive_kind']}:{sha[:16]}",
"access_class": "sensitive", "sha256": sha}
doc = engine.file_document_bytes(person_id, f"upload:{sha[:16]}", document_type, filename, content_type, content,
S.actor(caller), at, expires_on=expires_on)
return {"person_id": person_id, "doc_id": doc.doc_id, "access_class": doc.access_class, "sha256": doc.sha256}
@HR.use_case(kind="query", route="GET /people/{person_id}/documents", permission=S.DOC_READ)
def list_documents(caller: dict, person_id: BigId) -> dict:
"""A person's documents of the classes you may read: type, file name, the stored object's key, sha256, size,
who uploaded it and when, and its expiry. Sensitive documents only with hr:sensitive:read."""
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller)
p = view.person(person_id)
out = [{"doc_id": d.doc_id, "document_type": d.document_type, "access_class": d.access_class,
"filename": d.filename, "object_key": d.object_key, "sha256": d.sha256,
"size_bytes": d.size_bytes, "uploaded_by": d.uploaded_by, "uploaded_at": S.iso(d.uploaded_at),
"expires_on": S.due_label(d.expires_on), "contract_id": d.contract_id}
for d in sorted(p.documents.values(), key=lambda d: d.doc_id) if _may_read(caller, d.access_class)]
if S.has(caller, S.SENSITIVE_READ):
s = uow.sensitive.get(person_id)
for kind, t in sorted((s.kinds if s else {}).items()):
for sl in t.active():
if sl.data.get("object_key"):
v = sl.data.get("value") or {}
out.append({"doc_id": f"sensitive:{kind}:{(sl.data.get('sha256') or '')[:16]}",
"document_type": kind, "access_class": "sensitive",
"filename": v.get("filename"), "object_key": sl.data["object_key"],
"sha256": sl.data.get("sha256"), "size_bytes": v.get("size_bytes"),
"uploaded_by": sl.recorded_by, "uploaded_at": S.iso(sl.recorded_at),
"expires_on": v.get("expires_on"), "contract_id": None})
return {"person_id": person_id, "documents": out}
@HR.use_case(kind="query", route="GET /people/{person_id}/documents/{doc_id}")
def download_document(caller: dict, person_id: BigId, doc_id: Name) -> dict:
"""Stream one document's bytes (base64) after its class check — never an object link. A document of a class
you may not read (a manager asking for a right-to-work copy) is not found, through both doors alike."""
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller)
if not (S.has(caller, S.DOC_READ) or S.has(caller, S.PAY_READ) or S.has(caller, S.SENSITIVE_READ)):
raise Refused(404, f"no document {doc_id}")
p = view.person(person_id)
if doc_id.startswith("sensitive:"):
if not S.has(caller, S.SENSITIVE_READ):
raise Refused(404, f"no document {doc_id}")
_, kind, short = (doc_id.split(":") + ["", ""])[:3]
s = uow.sensitive.get(person_id)
sl = next((x for x in (s.kinds.get(kind).active() if s and s.kinds.get(kind) else [])
if (x.data.get("sha256") or "").startswith(short) and short), None)
if sl is None:
raise Refused(404, f"no document {doc_id}")
key, filename = sl.data["object_key"], (sl.data.get("value") or {}).get("filename") or kind
else:
d = p.documents.get(doc_id)
if d is None or not _may_read(caller, d.access_class):
raise Refused(404, f"no document {doc_id}")
key, filename = d.object_key, d.filename
got = uow.objects.get(key)
if got is None:
raise Refused(404, f"no document {doc_id}")
content, ctype, sha = got
return {"person_id": person_id, "doc_id": doc_id, "filename": filename, "content_type": ctype,
"sha256": sha, "size": len(content), "content_base64": base64.b64encode(content).decode()}
__all__ = ["file_document", "list_documents", "download_document"]