dodil/erp-brick-expenses
DODIL ERP brick: expenses
src/brick_expenses/application/migration.py
"""The Odoo slice (EXP-24): Odoo 17's posted and paid expense sheets as read-only claims, the unpaid ones as claims still to be paid here, the receipt images as
objects - reconciled per employee, per month and per image to Odoo, then signed by the owner (workflow exp_migration_run).
The load is READ-ONLY on Odoo's side (a reader port, infrastructure/odoo.py) and idempotent here: every claim is keyed odoo:<db>:hr.expense.sheet:<id> and
recorded in exp_migration_map, so a second load writes nothing new. History is never re-priced: a sheet's amounts are Odoo's own, to the penny. A SIGNED run is
frozen. The load posts NO journal for a sheet Odoo already posted (gl's migration carries Odoo's entries); a sheet Odoo only approved is posted by the normal
path from a pending row, and an unpaid claim asks payments for its payout the same way - neither is sent inside the load's transaction."""
import base64
import binascii
import json
from contextlib import contextmanager
from datetime import date
from typing import Annotated, Literal
from erp_platform.bricks import Refused
from erp_platform.types import BigId, text
from pydantic import Field
from brick_expenses.application import use_cases as hub
from brick_expenses.application.inputs import Code, Day, Reason
from brick_expenses.domain import odoo as od
from brick_expenses.domain import receipts as rc
from brick_expenses.domain import report as rp
from brick_expenses.domain.errors import RuleBroken
from brick_expenses.domain.money import ZERO
from brick_expenses.infrastructure import objects, seams
from brick_expenses.infrastructure import odoo as odoo_port
EXP = hub.EXP
MIG_LOAD, MIG_RECONCILE, MIG_READ, MIG_SIGN = "exp:migration:load", "exp:migration:reconcile", "exp:migration:read", "exp:migration:sign"
CHECKS = ("unpaid_reimbursable_per_employee", "claims_and_totals_per_month", "receipt_images_carried")
@contextmanager
def _odoo():
"""The reader, or the reply that there is none (409) / that Odoo failed (502)."""
try:
yield odoo_port.ports.get()
except odoo_port.OdooUnavailable as down:
raise Refused(down.http, down.reason) from None
def _run_view(run: dict) -> dict:
checks = json.loads(run["checks_json"] or "[]")
return {"run_id": run["run_id"], "odoo_db": run["odoo_db"], "cutover_date": run["cutover_date"].isoformat(), "company_code": run["company_code"],
"status": run["status"], "doors": json.loads(run["doors_json"] or "{}"), "checks": checks, "summary": json.loads(run["summary_json"] or "{}"),
"loaded_by": run["loaded_by"], "signed_by": run["signed_by"], "signed_on": run["signed_on"].isoformat() if run["signed_on"] else None,
"discard_reason": run["discard_reason"], "readiness": sign_gates(checks) if run["status"] == "reconciled" else None}
def sign_gates(checks: list) -> list:
"""The gate of sign_exp_reconciliation as {gate, met, reason, fix}: every check equal, or explained by the owner."""
open_ = [c["check"] for c in checks if not c["equal"] and not c.get("explained")]
must = [c["check"] for c in checks if not c["equal"] and c.get("must") == "equal" and not c.get("explained")]
return [{"gate": "checks_reconciled", "met": not open_ and bool(checks), "reason": ("checks neither equal nor explained: " + "; ".join(open_)) if open_ else
("the run has not been reconciled" if not checks else ""), "fix": "fix the mapping and load again, or explain each difference"},
{"gate": "must_equal", "met": not must, "reason": ("a check that must be equal differs: " + "; ".join(must)) if must else "", "fix": "explain it or load again"}]
def _run_id(repo, db: str, cutover: date) -> str:
base = f"odoo:{db}:{cutover.isoformat()}"
n, rid = 1, base
while (run := repo.migration.run(rid)) is not None and run["status"] == "discarded":
n += 1
rid = f"{base}:{n}"
return rid
def _need_run(repo, run_id: str) -> dict:
run = repo.migration.run(run_id)
if run is None:
raise Refused(404, f"no migration run {run_id}")
return run
# ── load ──────────────────────────────────────────────────────────────────────────────────────
@EXP.use_case(kind="command", route="POST /migration/odoo/load", permission=MIG_LOAD, idempotent=True)
def load_exp_from_odoo(caller: dict, odoo_db: Code, cutover_date: Day, company_code: Code, years: Annotated[list[Annotated[int, Field(ge=2000, le=2100)]],
Field(min_length=1, max_length=6)], employee_map: Annotated[dict[Annotated[str, Field(max_length=20)], BigId], Field(max_length=20000)],
product_purposes: Annotated[dict[Annotated[str, Field(max_length=64)], text(64, min_length=1)], Field(max_length=2000)],
upstream_signed: bool, drafts: Literal["finished_in_odoo", "re_enter"], images: bool = True) -> dict:
"""Load Odoo's expenses READ-ONLY: the sheets in state approve, post and done dated in `years` become claims keyed odoo:<db>:hr.expense.sheet:<id> (done is
paid, post is posted, and a sheet unpaid at cut-over is still to be paid here), their lines (payment_mode company_account is a card line), the categories of the
expensed products (category_code = the product's default_code, posting to the gl purpose named in product_purposes) and the receipt images as objects. The
doors are answered up front: employee_map (Odoo employee id -> hr person_id: hr offers no migration map), product_purposes, upstream_signed (the hr,
business-partner and gl migrations are signed: attested, no route says so), drafts (draft and submitted sheets are finished in Odoo, or re-entered - never
carried). Idempotent on the Odoo key: a second load writes nothing new; a signed run is never rewritten."""
if not upstream_signed:
raise Refused(409, "the hr, business-partner and gl migrations must be signed first: Odoo's people, partners and journals are theirs", gate="upstream_signed",
fix="sign them, then load")
cutover = hub._date(cutover_date, "cutover_date")
who = hub._who(caller)
with _odoo() as reader, hub._decide():
currency = hub._cur(reader.company_currency(odoo_db))
sheets = [x for x in reader.sheets(odoo_db, list(years)) if x.get("state") in od.CARRIED]
sheet_ids = [int(x["id"]) for x in sheets]
expenses = reader.expenses(odoo_db, sheet_ids)
by_sheet: dict = {}
for e in expenses:
by_sheet.setdefault(od.ref_id(e.get("sheet_id")), []).append(e)
products = {int(p["id"]): p for p in reader.products(odoo_db)}
problems = []
people = {od.ref_id(x.get("employee_id")) for x in sheets}
problems += [f"Odoo employee {e} has no person in employee_map" for e in sorted(p for p in people if p is not None and str(p) not in employee_map)]
used = {od.ref_id(e.get("product_id")) for e in expenses}
for pid in sorted(u for u in used if u is not None):
code = (products.get(pid) or {}).get("default_code")
if not code or code not in product_purposes:
problems.append(f"Odoo product {pid} ({(products.get(pid) or {}).get('name') or 'unknown'}) has no category: name its default_code and gl purpose in product_purposes")
problems += [f"sheet {x['id']} has no lines" for x in sheets if not by_sheet.get(int(x["id"]))]
if problems:
raise Refused(409, "The doors are not answered for everything Odoo holds: " + "; ".join(problems[:20]) + (f" (and {len(problems) - 20} more)" if len(problems) > 20 else ""),
gate="doors_decided", fix="complete employee_map and product_purposes, then load again")
attachments = reader.attachments(odoo_db, [int(e["id"]) for e in expenses]) if images else []
with hub.repository() as repo:
rid = _run_id(repo, odoo_db, cutover)
run = repo.migration.run(rid)
if run is not None and run["status"] == "signed":
raise Refused(409, f"run {rid} is signed: signed history is never rewritten - start a new run for what changed after the cut-over", gate="not_signed")
have = repo.migration.mapped(odoo_db, "hr.expense.sheet", sheet_ids)
for sid, row in have.items():
prior = repo.migration.run(row["run_id"])
if prior and prior["status"] == "signed" and row["run_id"] != rid:
raise Refused(409, f"sheet {sid} was carried by the signed run {row['run_id']}: it is not loaded again", gate="not_signed")
new = [x for x in sheets if str(x["id"]) not in have]
persons = {}
for e_id in {od.ref_id(x.get("employee_id")) for x in new}:
pid = int(employee_map[str(e_id)])
emp = seams.employment(pid, cutover, who)
persons[e_id] = (pid, emp.get("bp_id"), emp.get("cost_centre") or "")
cats = {}
for pid, p in products.items():
if p.get("default_code") in product_purposes and pid in used:
cats[pid] = p
now = hub._now()
existing = {c["category_code"] for c in repo.categories.all_in_force(company_code, cutover)}
added_cats = []
for p in cats.values():
if p["default_code"] not in existing and p["default_code"] not in {c["default_code"] for c in added_cats}:
repo.categories.put(company_code, p["default_code"], str(p.get("name") or p["default_code"])[:100], product_purposes[p["default_code"]], "standard",
date(2000, 1, 1), now, True, "receipt")
added_cats.append(p)
at_att: dict = {}
for a in attachments:
at_att.setdefault(int(a["res_id"]), []).append(a)
reports, map_rows, stats = [], [], {"sheets": len(sheets), "claims_written": 0, "claims_unchanged": len(sheets) - len(new), "lines": 0, "images": 0,
"categories_added": len(added_cats), "unpayable": []}
outbox = []
for x in new:
sid = int(x["id"])
person_id, bp_id, centre = persons[od.ref_id(x.get("employee_id"))]
day = od.sheet_day(x)
status = od.status_of(x["state"], x.get("payment_state"))
r = rp.Report(od.report_id(odoo_db, sid), person_id, company_code, bp_id, rp.Identity("", ""), currency, str(x.get("name") or "")[:200], status=status,
submitted_on=day, approved_on=day, paid_on=day if status == "paid" else None, payment_id=f"odoo:{odoo_db}:paid" if status == "paid" else None)
for n, e in enumerate(sorted(by_sheet[sid], key=lambda e: int(e["id"])), start=1):
code = products[od.ref_id(e.get("product_id"))]["default_code"]
cur_name = od.ref_name(e.get("currency_id")) or currency
r.lines.append(rp.Line(n, code, od.iso(e.get("date")) or day, str(e.get("name") or "")[:200], od.amount(e.get("total_amount"), "total_amount"),
ZERO, "card" if e.get("payment_mode") == "company_account" else "own", "receipt", centre, "", str(e.get("description") or "")[:500],
cur_name, od.amount(e.get("total_amount_currency"), "total_amount_currency"),
calc={"odoo": {"model": "hr.expense", "id": int(e["id"]), "sheet": sid}}))
map_rows.append({"odoo_db": odoo_db, "odoo_model": "hr.expense", "odoo_id": str(e["id"]), "ours": f"{r.report_id}:1:{n}", "run_id": rid})
for a in at_att.get(int(e["id"]), []):
try:
body = base64.b64decode(a.get("datas") or "", validate=True)
kind = a.get("mimetype") or ""
body = rc.decode(kind, base64.b64encode(body).decode())
except (binascii.Error, ValueError, RuleBroken):
stats.setdefault("images_skipped", []).append(f"attachment {a.get('id')}: not a receipt image")
continue
digest = rc.sha256(body)
key = rc.object_key(r.report_id, n, digest, kind)
objects.ports.receipts().put(key, body, kind)
if repo.receipts.get(key) is None:
repo.receipts.add({"object_key": key, "report_id": r.report_id, "line_no": n, "sha256": digest, "content_type": kind, "size_bytes": len(body),
"uploaded_by": who, "uploaded_at": hub._utc_naive(now), "retain_until": rc.retain_until(od.iso(e.get("date")) or day, 6),
"possible_duplicate_of": None}, now)
r.lines[-1].receipt_key = key
map_rows.append({"odoo_db": odoo_db, "odoo_model": "ir.attachment", "odoo_id": str(a["id"]), "ours": key, "run_id": rid})
stats["images"] += 1
if sum((ln.amount for ln in r.lines), ZERO) != od.amount(x.get("total_amount"), "total_amount"):
raise Refused(409, f"sheet {sid}: its lines add up to {r.total():.2f} but Odoo says {od.amount(x.get('total_amount'), 'total_amount'):.2f}", gate="sheet_total")
r.total_amount, r.reimbursable_amount = r.total(), r.reimbursable()
map_rows.append({"odoo_db": odoo_db, "odoo_model": "hr.expense.sheet", "odoo_id": str(sid), "ours": r.report_id, "run_id": rid})
if status == "approved": # Odoo booked nothing: the normal path posts it, then it is paid here
ref, body = hub._approval_journal(repo, r, cutover)
r.journal_id = int(body["journal_id"])
outbox.append(("gl", f"{ref}:approve", r.report_id, body))
if status in ("approved", "posted") and r.reimbursable() > ZERO:
if not r.bp_id:
stats["unpayable"].append(f"{r.report_id}: hr gives person {person_id} no person partner to pay")
elif status == "posted":
r.payout_ref = f"exp:{r.report_id}:r1:pay"
outbox.append(("payments", r.payout_ref, r.report_id, hub._payout_body(r, cutover)))
reports.append(r)
stats["lines"] += len(r.lines)
stats["claims_written"] = len(reports)
repo.migration.add_map(map_rows, now) if map_rows else None
repo.migration.add_reports(reports, now)
for target, ref, report_id, body in outbox:
repo.outbox.add(target, ref, report_id, body, now)
doors = {"employees": len(employee_map), "products": dict(product_purposes), "years": sorted(set(years)), "upstream_signed": True, "drafts": drafts, "images": images}
row = {"run_id": rid, "odoo_db": odoo_db, "cutover_date": cutover, "company_code": company_code, "status": "loaded", "doors_json": json.dumps(doors, sort_keys=True),
"checks_json": "[]", "summary_json": json.dumps(stats, sort_keys=True), "loaded_by": who, "signed_by": None, "signed_sub": None, "signed_on": None,
"discarded_by": None, "discard_reason": None}
if run is None:
repo.migration.add_run(row, now)
elif reports:
repo.migration.save_run({**run, "status": "loaded", "checks_json": "[]", "summary_json": json.dumps(stats, sort_keys=True)}, now) # new rows: reconcile again
row = repo.migration.run(rid)
else:
row = run
refs = {(t, ref) for t, ref, _, _ in outbox}
view = _run_view({**row, "summary_json": json.dumps(stats, sort_keys=True)} if run is None or reports else row)
if refs:
hub._drain(None, who) # everything pending: a journal that posts writes its payout row, sent on the next pass of the same drain
return view | {"stats": stats}
# ── reconcile, explain, sign, discard ─────────────────────────────────────────────────────────────
@EXP.use_case(kind="command", route="POST /migration/odoo/{run_id}/reconcile", permission=MIG_RECONCILE, idempotent=True)
def reconcile_exp_migration(caller: dict, run_id: Code, employee_map: Annotated[dict[Annotated[str, Field(max_length=20)], BigId], Field(max_length=20000)]) -> dict:
"""Compare what was loaded with Odoo, read again, and keep the checks on the run (loaded -> reconciled): the unpaid reimbursable amount per employee (Odoo's
own_account lines of sheets approved or posted and not paid against the claims here still to be paid), the claims and totals per month, and the receipt images
carried against Odoo's attachments. Each check says Odoo's number, ours and whether they are equal; one that must be equal and is not has to be fixed or
explained before the owner signs. employee_map is the same door the load used. A signed run is not touched."""
with _odoo() as reader, hub._decide():
with hub.repository() as repo:
run = _need_run(repo, run_id)
if run["status"] not in ("loaded", "reconciled"):
raise Refused(409, f"run {run_id} is {run['status']}: only a loaded run is reconciled")
db, doors = run["odoo_db"], json.loads(run["doors_json"])
sheets = [x for x in reader.sheets(db, doors["years"]) if x.get("state") in od.CARRIED]
expenses = reader.expenses(db, [int(x["id"]) for x in sheets])
n_att = reader.attachment_count(db, [int(e["id"]) for e in expenses]) if doors.get("images") else 0
by_sheet: dict = {}
for e in expenses:
by_sheet.setdefault(od.ref_id(e.get("sheet_id")), []).append(e)
mine = {m["odoo_id"]: m["ours"] for m in repo.migration.of_run(run_id, "hr.expense.sheet")}
claims = {rid: repo.reports.get(rid) for rid in mine.values()}
person_of = {int(e): int(p) for e, p in employee_map.items()}
# 1. unpaid reimbursable per employee (compared by Odoo employee)
odoo_unpaid, ours_unpaid = {}, {}
for x in sheets:
if x["state"] in ("approve", "post") and x.get("payment_state") != "paid":
e = od.ref_id(x.get("employee_id"))
odoo_unpaid[e] = odoo_unpaid.get(e, ZERO) + od.own_total(by_sheet.get(int(x["id"]), []))
emp_of_person = {p: e for e, p in person_of.items()}
for sid, rid in mine.items():
r = claims[rid]
if r is not None and r.status in ("approved", "posted"):
e = emp_of_person.get(r.person_id, f"person {r.person_id}")
ours_unpaid[e] = ours_unpaid.get(e, ZERO) + r.reimbursable()
per_emp = {str(e): {"odoo": f"{odoo_unpaid.get(e, ZERO):.2f}", "ours": f"{ours_unpaid.get(e, ZERO):.2f}"} for e in sorted({*odoo_unpaid, *ours_unpaid}, key=str)}
c1 = {"check": CHECKS[0], "must": "equal", "odoo": f"{sum(odoo_unpaid.values(), ZERO):.2f}", "ours": f"{sum(ours_unpaid.values(), ZERO):.2f}",
"detail": per_emp, "equal": all(v["odoo"] == v["ours"] for v in per_emp.values())}
# 2. claims and totals per month
odoo_m: dict = {}
for x in sheets:
m = od.month_of(od.sheet_day(x))
n, t = odoo_m.get(m, (0, ZERO))
odoo_m[m] = (n + 1, t + od.amount(x.get("total_amount"), "total_amount"))
ours_m: dict = {}
for rid in mine.values():
r = claims[rid]
if r is not None:
m = od.month_of(r.approved_on)
n, t = ours_m.get(m, (0, ZERO))
ours_m[m] = (n + 1, t + r.total())
months = {m: {"odoo": f"{odoo_m.get(m, (0, ZERO))[0]} / {odoo_m.get(m, (0, ZERO))[1]:.2f}", "ours": f"{ours_m.get(m, (0, ZERO))[0]} / {ours_m.get(m, (0, ZERO))[1]:.2f}"}
for m in sorted({*odoo_m, *ours_m})}
c2 = {"check": CHECKS[1], "must": "equal", "odoo": f"{sum(n for n, _ in odoo_m.values())} / {sum((t for _, t in odoo_m.values()), ZERO):.2f}",
"ours": f"{sum(n for n, _ in ours_m.values())} / {sum((t for _, t in ours_m.values()), ZERO):.2f}", "detail": months,
"equal": all(v["odoo"] == v["ours"] for v in months.values())}
# 3. images carried
carried = len([m for m in repo.migration.of_run(run_id, "ir.attachment")])
c3 = {"check": CHECKS[2], "must": "equal", "odoo": str(n_att), "ours": str(carried), "detail": {}, "equal": n_att == carried}
old = {c["check"]: c for c in json.loads(run["checks_json"] or "[]")}
checks = [{**c, "explained": (old.get(c["check"]) or {}).get("explained") if not c["equal"] else None} for c in (c1, c2, c3)]
row = {**run, "status": "reconciled", "checks_json": json.dumps(checks, sort_keys=True)}
repo.migration.save_run(row, hub._now())
return _run_view(row)
@EXP.use_case(kind="command", route="POST /migration/odoo/{run_id}/explain", permission=MIG_SIGN, idempotent=True)
def explain_exp_check(caller: dict, run_id: Code, check: Literal["unpaid_reimbursable_per_employee", "claims_and_totals_per_month", "receipt_images_carried"],
reason: Reason) -> dict:
"""The owner explains one difference the reconciliation found (a sheet Odoo still shows approved that was paid by hand, an attachment that is not an image),
so that it can be signed with it. A check that is equal needs no explanation; the reason is kept on the run."""
if not reason.strip():
raise Refused(422, "an explanation needs a reason", gate="bounds")
with hub.repository() as repo, hub._decide():
run = _need_run(repo, run_id)
if run["status"] != "reconciled":
raise Refused(409, f"run {run_id} is {run['status']}: only a reconciled run has differences to explain")
checks = json.loads(run["checks_json"])
c = next(x for x in checks if x["check"] == check)
if c["equal"]:
raise Refused(409, f"{check} is equal: there is nothing to explain")
c["explained"] = f"{reason.strip()} ({hub._who(caller)}, {hub._today(repo)})"
row = {**run, "checks_json": json.dumps(checks, sort_keys=True)}
repo.migration.save_run(row, hub._now())
return _run_view(row)
@EXP.use_case(kind="command", route="POST /migration/odoo/{run_id}/sign", permission=MIG_SIGN, idempotent=True)
def sign_exp_reconciliation(caller: dict, run_id: Code) -> dict:
"""The owner signs the reconciliation (reconciled -> signed, terminal): that nobody is owed more or less than Odoo says, every check equal or explained. A
signed run is frozen - a later load never rewrites it. Signing twice is one signature."""
with hub.repository() as repo, hub._decide():
run = _need_run(repo, run_id)
if run["status"] == "signed":
return _run_view(run)
if run["status"] != "reconciled":
raise Refused(409, f"run {run_id} is {run['status']}: only a reconciled run is signed", gate="checks_reconciled", fix="reconcile it first")
gates = sign_gates(json.loads(run["checks_json"]))
unmet = [g for g in gates if not g["met"]]
if unmet:
raise Refused(409, "Cannot sign: " + "; ".join(f"{g['gate']}: {g['reason']}" for g in unmet), gate=unmet[0]["gate"], fix=unmet[0]["fix"], fields={"gates": gates})
me = hub._me(caller)
repo.claims.once("migration_sign", run_id, hub._now(), hub._who(caller))
row = {**run, "status": "signed", "signed_by": me.email or me.sub, "signed_sub": me.sub, "signed_on": hub._today(repo)}
repo.migration.save_run(row, hub._now())
return _run_view(row)
@EXP.use_case(kind="command", route="POST /migration/odoo/{run_id}/discard", permission=MIG_SIGN, destructive=True)
def discard_exp_migration(caller: dict, run_id: Code, reason: Reason) -> dict:
"""The owner discards a run that is loaded or reconciled (a reason is required; terminal): the claims, lines, receipt rows and map it wrote are taken back
out. Refused once anything it wrote has left for a neighbour (a journal posted, a payout asked for) - then fix it and load again instead - and for a
signed run, which is history. The objects stay in the store: they are the same bytes under the same keys."""
if not reason.strip():
raise Refused(422, "a reason is required to discard a run", gate="bounds")
with hub.repository() as repo, hub._decide():
run = _need_run(repo, run_id)
if run["status"] not in ("loaded", "reconciled"):
raise Refused(409, f"run {run_id} is {run['status']}: only a loaded or reconciled run is discarded")
ids = [m["ours"] for m in repo.migration.of_run(run_id, "hr.expense.sheet")]
gone = [b for rid in ids for b in repo.outbox.of_report(rid) if b["status"] == "sent"]
if gone:
raise Refused(409, f"{len(gone)} of this run's journals or payouts have already left for gl or payments: it cannot be discarded", gate="nothing_sent",
fix="reconcile and fix instead")
moved = [rid for rid in ids if (c := repo.reports.get(rid)) is not None and c.status == "paid" and not str(c.payment_id or "").startswith("odoo:")]
if moved:
raise Refused(409, f"{len(moved)} claims were paid here since the load: it cannot be discarded", gate="nothing_sent")
repo.migration.delete_run(run_id, ids)
row = {**run, "status": "discarded", "discarded_by": hub._who(caller), "discard_reason": reason.strip()}
repo.migration.save_run(row, hub._now())
return _run_view(row)
@EXP.use_case(kind="query", route="GET /migration/odoo/{run_id}", permission=MIG_READ)
def get_exp_migration_run(caller: dict, run_id: Code) -> dict:
"""One migration run: its status, the doors it was loaded with, the reconciliation's checks (Odoo's number, ours, explained) and what signing still needs."""
with hub.repository() as repo:
return _run_view(_need_run(repo, run_id))
@EXP.use_case(kind="query", route="GET /migration/odoo", permission=MIG_READ)
def list_exp_migration_runs(caller: dict) -> dict:
"""Every migration run, oldest first: id, Odoo database, cut-over, status."""
with hub.repository() as repo:
return {"runs": [{"run_id": r["run_id"], "odoo_db": r["odoo_db"], "cutover_date": r["cutover_date"].isoformat(), "status": r["status"]} for r in repo.migration.runs()]}
@EXP.use_case(kind="query", route="GET /migration-map/{db}/{model}", permission=MIG_READ)
def get_exp_migration_map(caller: dict, db: Code, model: Literal["hr.expense.sheet", "hr.expense", "ir.attachment"],
ids: Annotated[str, Field(max_length=8000)]) -> list:
"""The one map route every migrating brick offers (shared.yaml conventions.odoo_migration): for a comma-separated list of Odoo ids, [{odoo_model, odoo_id,
ours}] - our key (a report id, a line key, an object key). An id not migrated is omitted, never null and never an error, so the caller sees exactly what
is missing."""
want = [i.strip() for i in ids.split(",") if i.strip()]
if not want or not all(i.isdigit() and len(i) <= 18 for i in want):
raise hub._bounds("ids are Odoo ids, comma-separated digits such as 12,13")
with hub.repository() as repo:
got = repo.migration.mapped(db, model, want)
return [{"odoo_model": model, "odoo_id": i, "ours": got[i]["ours"]} for i in want if i in got]