dodil/erp-brick-gl
DODIL ERP brick: general ledger
src/brick_gl/application/brick.py
"""The brick itself and what every use case module shares: the repository, the refusal, who the caller is, business time, the
companies a caller may see, and the lookups a posting needs.
Two things are decided in the use cases and nowhere else: WHO MAY (a PERMISSION on the declaration, never a role's name) and
WHICH ROWS (the companies the caller's gl_company_access rows name - a company outside them is a 404, never a 403).
"""
import functools
import re
from contextlib import contextmanager
from datetime import date, datetime, timezone
from typing import Optional
from erp_platform import auth, work
from erp_platform.bricks import Brick, Refused
from sqlalchemy.exc import IntegrityError
from brick_gl.domain import dates
from brick_gl.domain.errors import RuleBroken
from brick_gl.domain.journal import AccountInfo, CompanyInfo, Lookups
from brick_gl.infrastructure import business_time
# ── the permissions (module.yaml permissions.catalog) ────────────────────────────────────────────────────────────────
JOURNAL_READ, JOURNAL_WRITE, JOURNAL_APPROVE = "gl:journal:read", "gl:journal:write", "gl:journal:approve"
REVERSAL_APPROVE, ACCOUNT_WRITE = "gl:reversal:approve", "gl:account:write"
COMPANY_READ, COMPANY_WRITE, FSV_WRITE = "gl:company:read", "gl:company:write", "gl:fsv:write"
PERIOD_READ, PERIOD_CLOSE, PERIOD_REOPEN, ADJUSTMENT_POST = "gl:period:read", "gl:period:close", "gl:period:reopen", "gl:adjustment:post"
FX_READ, FX_WRITE, REVAL_RUN, RECONCILE_RUN, REPORT_READ = "gl:fx:read", "gl:fx:write", "gl:reval:run", "gl:reconcile:run", "gl:report:read"
RECURRING_WRITE, RECURRING_APPROVE, RECURRING_RUN = "gl:recurring:write", "gl:recurring:approve", "gl:recurring:run"
ACCRUAL_WRITE, ACCRUAL_APPROVE, ACCRUAL_RUN = "gl:accrual:write", "gl:accrual:approve", "gl:accrual:run"
ALLOCATION_WRITE, ALLOCATION_APPROVE, ALLOCATION_SIMULATE, ALLOCATION_RUN = ("gl:allocation:write", "gl:allocation:approve",
"gl:allocation:simulate", "gl:allocation:run")
KEYFIGURE_WRITE, BUDGET_WRITE, BUDGET_APPROVE = "gl:keyfigure:write", "gl:budget:write", "gl:budget:approve"
CLOSE_MANAGE, CLOSE_TASK, CLOSE_SIGNOFF = "gl:close:manage", "gl:close:task", "gl:close:signoff"
AUDIT_EXPORT, AUDIT_VERIFY = "gl:audit:export", "gl:audit:verify"
SETTINGS_WRITE, CONTROLS = "gl:settings:write", "gl:settings:controls"
MIGRATION_READ, MIGRATION_SIGN = "gl:migration:read", "gl:migration:sign"
# machine permissions: held by a service client by an exact grant; no wildcard reaches them
POST, BALANCE_DERIVE, CARRY_DERIVE, RESPLIT_POST = "gl:journal:post", "gl:balance:derive", "gl:carry_forward:derive", "gl:resplit:post"
CLOSE_CHECK, GROUP_EXTRACT, STATEMENT_MATERIALISE = "gl:close:check", "gl:group_data:extract", "gl:statement:materialise"
JOURNAL_CHECK, NUMBER_CLAIM, MIGRATION_LOAD, MIGRATION_RECONCILE = ("gl:journal:check", "gl:number_series:claim", "gl:migration:load",
"gl:migration:reconcile")
COST_CENTRE_READ = "gl:cost_centre:read"
MACHINE = (POST, BALANCE_DERIVE, CARRY_DERIVE, RESPLIT_POST, CLOSE_CHECK, GROUP_EXTRACT, STATEMENT_MATERIALISE, JOURNAL_CHECK,
NUMBER_CLAIM, MIGRATION_LOAD, MIGRATION_RECONCILE)
GL = Brick("gl", version="0.1.1", ids_as_strings=True, machine_permissions=MACHINE, instructions=(
"The books. Every transaction is a balanced journal, posted once: other modules post through post_from_source, "
"idempotent by source document and event (the same call again answers the journal already posted). A journal is never "
"edited or deleted: a correction is a reversing journal. Balances are derived from journal lines, never typed. A "
"closed period refuses operational postings; check_period_open says whether a date is open. Companies, ledgers "
"(IFRS beside local GAAP), profit centres and segments are dimensions of every line; a posting that names no company "
"lands in the default company and counts in every ledger. get_trial_balance proves the books; get_account_lines "
"traces any balance to its source. Controls on the people who keep the books are the owner's (set_gl_control)."))
GL.declare_process("process.yaml")
# ── the shape of a code, a period, a year or a date is checked HERE, after the permission ─────────────────────────────────────
_PATTERNS = (
(re.compile(r"^(period_id|[a-z]+_period)$"), re.compile(r"^[0-9]{4}-(0[1-9]|1[0-2])$"), "a period is YYYY-MM"),
(re.compile(r"^fiscal_year$"), re.compile(r"^FY[0-9]{4}$"), "a financial year is named FY2026"),
(re.compile(r"^(company_code|partner_company)$"), re.compile(r"^[A-Z0-9]{1,4}$"), "a company code is 1 to 4 upper-case letters or digits"),
(re.compile(r"^ledger_id$"), re.compile(r"^[A-Z0-9]{2,8}$"), "a ledger id is 2 to 8 upper-case letters or digits"),
(re.compile(r"^(currency|from_currency|to_currency|to)$"), re.compile(r"^[A-Z]{3}$"), "a currency is an ISO 4217 code such as GBP"),
)
_DATES = re.compile(r"^(journal_date|rate_date|date|as_at|as_of|key_date|date_from|date_to|valid_from|valid_until|active_until|reversal_date|"
r"cutover_date)$")
def _check_arg(name: str, value) -> None:
if value is None or not isinstance(value, str):
return
if _DATES.match(name):
try:
d = date.fromisoformat(value.strip()[:10])
ok = date(1900, 1, 1) <= d <= date(2199, 12, 31)
except ValueError:
ok = False
if not ok:
raise Refused(422, f"{name} is a date like 2026-11-01 (from 1900 to 2199), not {value!r}")
return
for key, pattern, words in _PATTERNS:
if key.match(name) and not pattern.match(value):
raise Refused(422, f"{words}, not {value!r} ({name})")
def _checked(fn):
@functools.wraps(fn)
def wrapper(caller, **kw):
for k, v in kw.items():
_check_arg(k, v)
return fn(caller, **kw)
return wrapper
_declare = GL.use_case
def _use_case(**kw):
register = _declare(**kw)
def decorate(fn):
register(_checked(fn))
return fn
return decorate
GL.use_case = _use_case
@contextmanager
def repository():
"""A unit of work: `use_cases.repository()` (tests and the scenario harness swap that one attribute, so every module of
the brick reaches storage through it)."""
from brick_gl.application import use_cases
with use_cases.repository() as repo:
yield repo
@contextmanager
def decide():
"""A broken business rule is a refusal with the rule's own words and status, through both doors."""
try:
yield
except RuleBroken as broken:
raise Refused(broken.status, str(broken)) from None
def who(caller: dict) -> str:
"""The caller as a stored author is: their email, else their sub."""
return (caller.get("email") or caller.get("sub") or "").strip()
def sub_of(caller: dict) -> str:
return (caller.get("sub") or caller.get("email") or "").strip()
def ids(caller: dict) -> tuple:
"""Every identity the caller has (email and sub, lower-case): the same person under either is the same person."""
return tuple(dict.fromkeys(x.strip().lower() for x in (caller.get("email"), caller.get("sub")) if x and x.strip()))
def same_person(caller: dict, *stored) -> bool:
mine = set(ids(caller))
return any(s and s.strip().lower() in mine for s in stored)
def real_now() -> datetime:
"""Audit stamps are real time (naive UTC); business time is `today()`."""
return datetime.now(timezone.utc).replace(tzinfo=None)
def business_now() -> datetime:
return work.now()
def today(repo) -> date:
"""The date NOW (work.now(): a demo may move the clock) in the bucket's business_timezone - ONE value for every brick
(shared.yaml conventions.business_timezone), read only here so it can move to the platform later. A dated row takes
effect once the business date in the zone BEFORE it has reached it; today never goes backwards."""
return business_time.resolve(repo.settings.history("business_timezone", ""), work.now())[1]
def business_timezone(repo) -> str:
return business_time.resolve(repo.settings.history("business_timezone", ""), work.now())[0]
def parse_day(v: Optional[str], what: str) -> Optional[date]:
try:
return dates.parse(v)
except ValueError:
raise Refused(422, f"{what} is a date like 2026-11-01") from None
# ── which companies the caller sees ──────────────────────────────────────────────────────────────────────────────────
def allowed_companies(repo, caller: dict) -> Optional[set]:
"""None = every company; else the codes the caller's company_access rows put in force today (GL-60)."""
return repo.access.allowed(sub_of(caller), (caller.get("email") or "").strip(), today(repo))
def can_see(repo, caller: dict, company_code: str) -> bool:
allowed = allowed_companies(repo, caller)
return allowed is None or company_code in allowed
def company_row(repo, caller: dict, company_code: Optional[str], *, missing: int = 404, scoped: bool = True) -> dict:
"""The named company, or the default one when none is named. A company the caller may not see is a 404, not a 403 (a 403
confirms it exists); an unknown code is 404 for a read and 422 company_known for a posting (`missing=422`)."""
row = repo.companies.get(company_code) if company_code else repo.companies.default()
if row is None:
if missing == 422:
raise Refused(422, f"[company_known] {'the company ' + company_code if company_code else 'a default company'} "
"is not a company of these books (fix: Companies: add it, or send the default)")
raise Refused(404, f"no company {company_code}" if company_code else "no company is set up yet")
if scoped and not can_see(repo, caller, row["company_code"]):
raise Refused(404, f"no company {row['company_code']}")
return row
def visible_codes(repo, caller: dict) -> list:
allowed = allowed_companies(repo, caller)
return [c["company_code"] for c in repo.companies.all() if allowed is None or c["company_code"] in allowed]
def leading_ledger(repo) -> str:
row = repo.ledgers.leading()
return row["ledger_id"] if row else ""
def ledger_for(repo, company_code: str, ledger_id: Optional[str], fiscal_year: int) -> str:
"""The ledger a read is run on: the one named (it must be assigned to the company that year), else the leading one."""
if not ledger_id:
return leading_ledger(repo)
if ledger_id not in repo.company_ledgers.assigned(company_code, fiscal_year):
raise Refused(422, f"[ledger_assigned] the ledger {ledger_id} is not assigned to {company_code} for FY{fiscal_year} "
"(fix: Companies: assign the ledger, or post to all ledgers)")
return ledger_id
def unique_violation(exc: BaseException) -> bool:
"""A lost claim (a plain INSERT of a key another transaction holds): SQLSTATE 23505 / UNIQUE failed."""
if not isinstance(exc, IntegrityError):
return False
text = (str(exc) + " " + str(getattr(exc, "orig", ""))).lower()
return "unique" in text or "duplicate key" in text or "23505" in text
def claims(unit, attempts: int = 8):
"""Run one whole unit of work (its own transaction) again when it lost a claim - a number, a chain link, a journal id
another writer took first. The retry reads the committed state: a journal that now exists is a replay."""
for attempt in range(1, attempts + 1):
try:
return unit()
except IntegrityError as exc:
if not unique_violation(exc) or attempt == attempts:
if unique_violation(exc):
raise Refused(409, "the books are busy with another posting for the same number: try again") from None
raise
raise AssertionError("unreachable")
# ── the lookups a posting needs, over the repository ─────────────────────────────────────────────────────────────────
class RepoLookups(Lookups):
def __init__(self, repo, company: dict):
self.repo = repo
self.company = CompanyInfo(company["company_code"], company["functional_currency"], company["chart_id"],
company.get("default_profit_centre") or "", company.get("bp_id"))
self._accounts: dict = {}
self._charts: dict = {}
self._partners: Optional[dict] = None
def _chart_kind(self, chart_id: str) -> str:
if chart_id not in self._charts:
row = self.repo.charts.get(chart_id)
self._charts[chart_id] = row["kind"] if row else "operational"
return self._charts[chart_id]
def prefetch(self, account_ids) -> None:
"""Read the accounts a journal names in ONE query (account() answers them from memory afterwards)."""
from brick_gl.infrastructure import tables as _t
ids = sorted({int(a) for a in account_ids if a is not None} - set(self._accounts))
if not ids:
return
found = {r["account_id"]: r for r in self.repo.accounts.rows(_t.gl_account.c.account_id.in_(ids))}
for a in ids:
self._accounts[a] = self._info(found.get(a))
def account(self, account_id: int) -> Optional[AccountInfo]:
if account_id not in self._accounts:
self._accounts[account_id] = self._info(self.repo.accounts.get(account_id))
return self._accounts[account_id]
def _info(self, r) -> Optional[AccountInfo]:
return None if r is None else AccountInfo(
account_id=r["account_id"], chart_id=r["chart_id"], code=r["code"], name=r["name"], type=r["type"],
subtype=r["subtype"] or "other", group_chart=self._chart_kind(r["chart_id"]) == "group",
source_only=bool(r["source_only"]), monetary=bool(r["monetary"]), currency=r["currency"],
default_profit_centre=r["default_profit_centre"] or "", fs_cost_centre=r["fs_cost_centre"] or "optional",
fs_profit_centre=r["fs_profit_centre"] or "optional", fs_bp_id=r["fs_bp_id"] or "optional",
fs_tax_code=r["fs_tax_code"] or "optional", active_until=r["active_until"])
def mapped(self, purpose: str, qualifier: str, on: date) -> Optional[int]:
k = (purpose, qualifier, on) # a purpose named on several lines is resolved once per journal
if k not in self.__dict__.setdefault("_mapped", {}):
self._mapped[k] = self.repo.account_map.account(self.company.chart_id, purpose, qualifier, on)
return self._mapped[k]
def cost_centre(self, code: str, on: date) -> Optional[dict]:
r = self.repo.dims.cost_centres.get(code)
if r is None or (r["company_code"] and r["company_code"] != self.company.company_code):
return None
closed = r["active_until"] is not None and r["active_until"] <= on
return {"status": "closed" if closed else "open", "company_code": r["company_code"],
"profit_centre": self.repo.dims.profit_centre_of_cost_centre(code, on)}
def profit_centre_valid(self, code: str, on: date) -> bool:
r = self.repo.dims.profit_centres.get(code)
if r is None:
return False
codes = [c for c in (r["company_codes"] or "").split(",") if c]
if codes and self.company.company_code not in codes:
return False
return (r["valid_from"] is None or r["valid_from"] <= on) and (r["valid_until"] is None or r["valid_until"] >= on)
def segment_of(self, profit_centre: str, on: date) -> str:
return self.repo.dims.segment_of(profit_centre, on) if profit_centre else ""
def company_of_partner(self, bp_id: int) -> Optional[str]:
if self._partners is None:
self._partners = {c["bp_id"]: c["company_code"] for c in self.repo.companies.all() if c["bp_id"] is not None}
return self._partners.get(int(bp_id))
# the bounded identity every use case needs from `auth` is imported here so a module has one place to look
require_permission = auth.has_permission
__all__ = ["GL", "Refused", "RuleBroken", "repository", "decide"]