dodil/erp-brick-fixed-assets
DODIL ERP brick: fixed assets
src/brick_fixed_assets/infrastructure/seams.py
"""The neighbours this brick asks, behind one adapter each. Declared HERE and nowhere else, so the domain never knows
how they are reached. Every route is the neighbour's SPEC route (module.yaml seams), exact:
gl POST /journal-entry/journals one journal per transaction (or run) and posting area
GET /companies/{company_code}/ledgers an area's ledger must be one of these
GET /periods/open the period_open gate per posting area (ledger_id = the multi-ledger extension)
GET /fx-rates/{currency} only an acquisition in a foreign currency (translated once, kept)
inventory GET /locations the location picker (optional: without it a site code is free text)
fixed-assets never writes gl_ tables; gl never reads fa_ tables (it asks GET /assets/balances and GET /depreciation/status).
Every answer is a value, never an exception, except `Unavailable` where the work cannot go on without the neighbour.
"""
import json
from dataclasses import dataclass
from datetime import date
from typing import Optional
from erp_platform.seam import Seam
GL = Seam("gl", "GL_BASE_URL", roles=["service"],
permissions=["gl:journal:post", "gl:period:read", "gl:company:read", "gl:fx:read"])
INVENTORY = Seam("inventory", "INVENTORY_BASE_URL", roles=["service"], permissions=["inv:stock:read"])
class Unavailable(Exception):
"""The neighbour is not installed, refused or did not answer: `http` is what this brick answers, `reason` says why."""
def __init__(self, reason: str, http: int = 503):
super().__init__(reason)
self.reason, self.http = reason, http
def _json(text: str) -> dict:
try:
v = json.loads(text)
return v if isinstance(v, dict) else {}
except (ValueError, TypeError):
return {}
@dataclass
class PeriodState:
open: Optional[bool] # None = could not be checked
period_id: Optional[str] = None
reason: str = ""
fix: str = ""
def period_open(day: date, company_code: str, ledger_id: Optional[str], actor: str = "") -> PeriodState:
"""gl's period gate for a date, in one company's ledger. Not installed, or no answer: open is None with the reason."""
if not GL.installed:
return PeriodState(None, reason="gl is not installed, so the period cannot be checked", fix="compose gl with fixed-assets")
params = {"date": day.isoformat(), "company_code": company_code}
if ledger_id is not None:
params["ledger_id"] = str(ledger_id)
r = GL.get("/periods/open", params, actor=actor)
if not r.ok:
return PeriodState(None, reason=f"gl answered {r.status or 'nothing'}: {r.reason}", fix="try again")
d = r.data
return PeriodState(bool(d.get("open")), d.get("period_id"), d.get("reason") or "", d.get("fix") or "")
def company_ledgers(company_code: str, actor: str = "") -> dict:
"""gl's ledgers of a company: {ledgers: [{ledger_id, kind, ...}], fiscal_year_end_month, functional_currency}.
Raises Unavailable when gl is not installed or does not know the company."""
if not GL.installed:
raise Unavailable("gl is not installed, so an area's ledger cannot be checked", 409)
r = GL.get(f"/companies/{company_code}/ledgers", actor=actor)
if r.status == 404:
raise Unavailable(f"gl does not know company {company_code}", 422)
if not r.ok:
raise Unavailable(f"gl answered {r.status or 'nothing'} ({r.reason})", 502 if r.status else 503)
return r.data
def fiscal_year_end_month(company_code: str) -> int:
"""The company's fiscal year end month (gl's), 12 when gl is not there to say."""
try:
return int(company_ledgers(company_code).get("fiscal_year_end_month") or 12)
except Unavailable:
return 12
@dataclass
class Posted:
ok: bool
journal_id: Optional[int] = None
replay: bool = False
status: int = 0
gate: str = ""
reason: str = ""
fix: str = ""
def post_journal(body: dict, actor: str = "") -> Posted:
"""POST /journal-entry/journals. 201 posted, 200 the same replay; 409 {gate: period_open | journal_conflict} and
422 {gate: balanced | purposes_mapped | ledger_assigned | bounds} come back as a refusal with gl's own words."""
if not GL.installed:
raise Unavailable("gl is not installed: nothing can post (gl is required)", 409)
r = GL.post("/journal-entry/journals", body, actor=actor)
if r.ok:
return Posted(True, int(r.data.get("journal_id") or body["journal_id"]), r.status == 200 or bool(r.data.get("replay")),
r.status)
if r.status in (409, 422):
d = _json(r.reason)
return Posted(False, None, False, r.status, d.get("gate") or "", d.get("reason") or r.reason, d.get("fix") or "")
raise Unavailable(f"gl answered {r.status or 'nothing'} ({r.reason}): the journal was not posted; try again", 502 if r.status else 503)
def fx_rate(currency: str, to: str, on: date, actor: str = "") -> Optional[dict]:
"""gl's rate for translating a foreign acquisition once at its value date; None when gl has none (404)."""
if not GL.installed:
raise Unavailable("gl is not installed, so a foreign currency cannot be translated", 409)
r = GL.get(f"/fx-rates/{currency}", {"on": on.isoformat(), "to": to, "type": "spot"}, actor=actor)
if r.status == 404:
return None
if not r.ok:
raise Unavailable(f"gl answered {r.status or 'nothing'} ({r.reason})", 502 if r.status else 503)
return r.data
def locations(actor: str = "") -> Optional[list]:
"""inventory's internal locations, or None when inventory is not composed (a site code is then free text)."""
if not INVENTORY.installed:
return None
r = INVENTORY.get("/locations", {"usage": "internal"}, actor=actor)
if not r.ok:
raise Unavailable(f"inventory answered {r.status or 'nothing'} ({r.reason})", 502 if r.status else 503)
return r.data.get("locations") or []