dodil/erp-brick-manufacturing

DODIL ERP brick: manufacturing

src/brick_manufacturing/infrastructure/odoo.py

"""The Odoo reader port (MFG-32): read-only access to an Odoo 17 mrp database behind one small interface, so the migration never knows how
Odoo is reached and tests hand it fake rows. `ports.reader` is what the use cases use; None refuses with a plain message until a composition
wires one. NO live reader is shipped here (the mapping is verified by fake rows, tests/test_migration.py; a reader over Odoo's external API
returns these shapes from the models the spec lists).

Rows are plain dicts with Odoo's meaning and our codes: `product` is the product code product's migration map resolves (the map is product's;
this brick never keeps one), quantities are numbers or numeric strings, dates ISO strings:

  workcentres(db)        [{id, name, code, costs_hour, capacity}]                                         mrp.workcenter (active)
  boms(db)               [{id, product, qty, uom, type (normal|phantom|subcontract), code,
                           lines: [{product, qty, uom}], byproducts: [{product, qty, uom}],
                           operations: [{id, name, workcenter_id, sequence, time_cycle_manual}]}]      mrp.bom with lines, byproducts, operations
  productions(db, years) [{id, name, state (draft|confirmed|progress|to_close|done|cancel), product, product_qty,
                           qty_produced, uom, date_finished, lot_producing, consumed: [{product, lot, qty}],
                           workorders: [{id, workcenter_id, sequence, duration}]}]                      mrp.production (+ move lines, workorders)
"""
from typing import Optional, Protocol


class OdooUnavailable(Exception):
    """No Odoo connection (409) or the call failed (502): a reply, never a half-loaded run."""

    def __init__(self, http: int, reason: str):
        super().__init__(reason)
        self.http, self.reason = http, reason


class OdooReader(Protocol):
    def workcentres(self, db: str) -> list: ...
    def boms(self, db: str) -> list: ...
    def productions(self, db: str, years: list) -> list: ...


class Ports:
    def __init__(self):
        self.reader: Optional[OdooReader] = None

    def get(self) -> OdooReader:
        if self.reader is None:
            raise OdooUnavailable(409, "no Odoo connection is wired: the migration reads Odoo read-only through a reader the composition provides")
        return self.reader


ports = Ports()