dodil/erp-brick-payments
DODIL ERP brick: payments
src/brick_payments/infrastructure/odoo.py
"""The Odoo migration's ports (PAY-44): a READER of the Odoo database (read-only, the operator's machine, test copy
first) and the MAP from an Odoo partner to our bp_id (business-partner's migration map). Nothing here writes Odoo.
The reader returns NORMALISED rows - flat dicts, many2one fields reduced to the id (or the code for a currency /
journal) - for the models of the spec's `migration.odoo.reads`:
account.journal id, code, name, type, currency, acc_number, gl_account
account.payment id, date, amount, currency, partner (commercial partner id), journal_code, payment_type
(inbound | outbound), partner_type, state, ref, method_code, move_id
account.bank.statement.line id, date, amount, currency, payment_ref, partner, journal_code, unique_import_id,
is_reconciled, amount_residual, move_id, payment_id (the account.payment behind it, or None)
account.move.line id, move_id, move_type, move_name, partner, date, date_maturity, amount_currency, currency,
amount_residual, amount_residual_currency, reconciled, payment_id, statement_line_id,
followup_level (Enterprise account_followup; 0 if none), followup_date
(receivable lines of posted moves only)
account.partial.reconcile id, debit_line, credit_line, amount, debit_amount_currency, credit_amount_currency,
debit_currency, credit_currency, max_date, full_reconcile_id, exchange_move_id
res.partner id, commercial_partner_id, credit_limit, use_partner_credit_limit
The follow-up fields are an addition to the spec's field list: the signed door decision carries each open invoice's
follow-up level, and Odoo keeps it on the receivable line. Tests use an in-memory fake; `JsonRpcReader` is the real
adapter and has not been exercised against a live Odoo.
"""
import os
from dataclasses import dataclass, field
from typing import Optional, Protocol
from brick_payments.infrastructure import seams
class OdooReader(Protocol):
db: str
company_currency: str
def read(self, model: str) -> list: ...
class PartnerMap(Protocol):
def resolve(self, odoo_partner_ids: list) -> dict:
"""odoo res.partner id -> our bp_id, for those the register knows."""
@dataclass
class Ports:
reader: Optional[OdooReader] = None
partners: Optional[PartnerMap] = None
_PORTS: Optional[Ports] = None
def use(ports: Optional[Ports]) -> None:
"""Install the ports (tests, or a composition's startup)."""
global _PORTS
_PORTS = ports
def ports() -> Ports:
"""The installed ports, else the real ones from the environment: ODOO_URL, ODOO_DB, ODOO_USER, ODOO_PASSWORD (a
READ-ONLY Odoo user) and the business-partner seam."""
if _PORTS is not None:
return _PORTS
reader = JsonRpcReader.from_env()
return Ports(reader, SeamPartnerMap(reader.db) if seams.PARTNERS.installed and reader else None)
class SeamPartnerMap:
"""business-partner's migration map, asked through its seam: GET /migration-map/{db}/res.partner?ids= ->
[{odoo_model, odoo_id, ours}] (shared.yaml conventions.odoo_migration; an unmapped id is left out)."""
def __init__(self, db: str = ""):
self.db = db
def resolve(self, odoo_partner_ids: list) -> dict:
out, ids = {}, list(odoo_partner_ids)
for i in range(0, len(ids), 200):
chunk = ids[i:i + 200]
r = seams.PARTNERS.get(f"/migration-map/{self.db}/res.partner", params={"ids": ",".join(str(p) for p in chunk)},
actor="odoo-migration")
want = {str(p): p for p in chunk}
for x in (r.data.get("body") or []) if r.ok else []:
if str(x.get("odoo_id")) in want and x.get("ours") not in (None, ""):
out[want[str(x["odoo_id"])]] = int(x["ours"])
return out
@dataclass
class JsonRpcReader:
"""Odoo's external API (JSON-RPC /jsonrpc, execute_kw search_read) as a READER. Not exercised against a live Odoo."""
url: str
db: str
login: str
password: str
company_currency: str = "GBP"
_uid: Optional[int] = field(default=None, repr=False)
@classmethod
def from_env(cls) -> Optional["JsonRpcReader"]:
env = {k: os.environ.get(k, "") for k in ("ODOO_URL", "ODOO_DB", "ODOO_USER", "ODOO_PASSWORD")}
if not all(env.values()):
return None
return cls(env["ODOO_URL"].rstrip("/"), env["ODOO_DB"], env["ODOO_USER"], env["ODOO_PASSWORD"],
os.environ.get("ODOO_COMPANY_CURRENCY", "GBP"))
def _call(self, service: str, method: str, *args):
import httpx
r = httpx.post(f"{self.url}/jsonrpc", json={"jsonrpc": "2.0", "method": "call", "id": 1,
"params": {"service": service, "method": method, "args": list(args)}}, timeout=120)
r.raise_for_status()
body = r.json()
if "error" in body:
raise RuntimeError(f"Odoo refused: {body['error'].get('data', {}).get('message', body['error'])}")
return body["result"]
def _search_read(self, model: str, domain: list, fields: list) -> list:
if self._uid is None:
self._uid = self._call("common", "login", self.db, self.login, self.password)
return self._call("object", "execute_kw", self.db, self._uid, self.password, model, "search_read", [domain], {"fields": fields})
@staticmethod
def _id(v):
return v[0] if isinstance(v, (list, tuple)) and v else (v or None)
@staticmethod
def _code(v):
return (v[1] if isinstance(v, (list, tuple)) and len(v) > 1 else "") or ""
def read(self, model: str) -> list:
sr, ident, code = self._search_read, self._id, self._code
if model == "account.journal":
return [{"id": r["id"], "code": r["code"], "name": r["name"], "type": r["type"], "currency": code(r["currency_id"]),
"acc_number": "", "gl_account": code(r["default_account_id"]).split(" ")[0]}
for r in sr(model, [["type", "in", ["bank", "cash"]]], ["code", "name", "type", "currency_id", "default_account_id"])]
partner = {p["id"]: ident(p["commercial_partner_id"]) for p in sr("res.partner", [], ["commercial_partner_id"])}
if model == "res.partner":
return [{"id": r["id"], "commercial_partner_id": ident(r["commercial_partner_id"]), "credit_limit": r["credit_limit"],
"use_partner_credit_limit": r["use_partner_credit_limit"]}
for r in sr(model, [["customer_rank", ">", 0], ["use_partner_credit_limit", "=", True]],
["commercial_partner_id", "credit_limit", "use_partner_credit_limit"])]
if model == "account.payment":
return [{"id": r["id"], "date": r["date"], "amount": r["amount"], "currency": code(r["currency_id"]),
"partner": partner.get(ident(r["partner_id"])), "journal_code": code(r["journal_id"]).split(" ")[0],
"payment_type": r["payment_type"], "partner_type": r["partner_type"], "state": r["state"], "ref": r["ref"] or "",
"method_code": code(r["payment_method_line_id"]), "move_id": ident(r["move_id"])}
for r in sr(model, [["partner_type", "=", "customer"], ["state", "in", ["posted", "paid"]]],
["date", "amount", "currency_id", "partner_id", "journal_id", "payment_type", "partner_type", "state",
"ref", "payment_method_line_id", "move_id"])]
if model == "account.bank.statement.line":
return [{"id": r["id"], "date": r["date"], "amount": r["amount"], "currency": code(r["currency_id"]),
"payment_ref": r["payment_ref"] or "", "partner": partner.get(ident(r["partner_id"])),
"journal_code": code(r["journal_id"]).split(" ")[0], "unique_import_id": r["unique_import_id"] or "",
"is_reconciled": r["is_reconciled"], "amount_residual": r["amount_residual"], "move_id": ident(r["move_id"]),
"payment_id": (r.get("payment_ids") or [None])[0]}
for r in sr(model, [["journal_id.type", "=", "bank"], ["amount", ">", 0], ["move_id.state", "=", "posted"]],
["date", "amount", "currency_id", "payment_ref", "partner_id", "journal_id", "unique_import_id",
"is_reconciled", "amount_residual", "move_id", "payment_ids"])]
if model == "account.move.line":
base = ["move_id", "move_type", "move_name", "partner_id", "date", "date_maturity", "amount_currency", "currency_id",
"amount_residual", "amount_residual_currency", "reconciled", "payment_id", "statement_line_id"]
domain = [["account_id.account_type", "=", "asset_receivable"], ["parent_state", "=", "posted"]]
try:
rows = sr(model, domain, base + ["followup_line_id", "followup_date"])
except RuntimeError: # no Enterprise account_followup: no level to carry
rows = sr(model, domain, base)
return [{"id": r["id"], "move_id": ident(r["move_id"]), "move_type": r["move_type"], "move_name": code(r["move_id"]),
"partner": partner.get(ident(r["partner_id"])), "date": r["date"], "date_maturity": r["date_maturity"],
"amount_currency": r["amount_currency"], "currency": code(r["currency_id"]),
"amount_residual": r["amount_residual"], "amount_residual_currency": r["amount_residual_currency"],
"reconciled": r["reconciled"], "payment_id": ident(r["payment_id"]),
"statement_line_id": ident(r["statement_line_id"]),
"followup_level": int(ident(r.get("followup_line_id")) or 0), "followup_date": r.get("followup_date") or None}
for r in rows]
if model == "account.partial.reconcile":
lines = {r["id"] for r in self.read("account.move.line")}
return [{"id": r["id"], "debit_line": ident(r["debit_move_id"]), "credit_line": ident(r["credit_move_id"]),
"amount": r["amount"], "debit_amount_currency": r["debit_amount_currency"],
"credit_amount_currency": r["credit_amount_currency"], "debit_currency": code(r["debit_currency_id"]),
"credit_currency": code(r["credit_currency_id"]), "max_date": r["max_date"],
"full_reconcile_id": ident(r["full_reconcile_id"]), "exchange_move_id": ident(r["exchange_move_id"])}
for r in sr(model, [], ["debit_move_id", "credit_move_id", "amount", "debit_amount_currency",
"credit_amount_currency", "debit_currency_id", "credit_currency_id", "max_date",
"full_reconcile_id", "exchange_move_id"])
if ident(r["debit_move_id"]) in lines and ident(r["credit_move_id"]) in lines]
if model == "account.full.reconcile":
return [{"id": r["id"]} for r in sr(model, [], ["id"])]
raise ValueError(f"{model} is not one of the models this migration reads")