dodil/erp-brick-gl
DODIL ERP brick: general ledger
src/brick_gl/application/masterdata.py
"""The master data every posting reads: companies, ledgers, charts and accounts, the account map, dimensions, settings and
the owner's controls, company access, and FX rates. This module is their sole writer (shared.yaml: gl owns them).
A thing a posting has USED is fixed (an account's code and type, a company's currency, a ledger's id); anything that
changes what a past day meant is a DATED row that starts after today, never an edit. Every change lands in gl_master_audit
with before and after.
"""
import re
from datetime import date
from typing import Optional
from erp_platform.bricks import Refused
from erp_platform.seam import derived_id
from erp_platform.types import BigId, text
from brick_gl.application.brick import (
ACCOUNT_WRITE,
COMPANY_READ,
COMPANY_WRITE,
CONTROLS,
COST_CENTRE_READ,
FX_READ,
FX_WRITE,
GL,
SETTINGS_WRITE,
allowed_companies,
company_row,
decide,
real_now,
repository,
sub_of,
today,
visible_codes,
who,
)
from brick_gl.application.inputs import (
Code,
CompanyCode,
Currency,
FiscalYearName,
FxRate,
IsoDate,
LedgerCode,
)
from brick_gl.domain import dates
from brick_gl.domain.errors import RuleBroken
from brick_gl.domain.values import (
ACCOUNT_TYPES,
CHART_KINDS,
COMPANY_CODE,
CONTROL_KEYS,
CURRENCY,
DEFAULT_LEDGER,
FIELD_STATUS,
LEDGER_CODE,
LEDGER_KINDS,
PURPOSES,
RATE_TYPES,
REVAL_METHODS,
SETTINGS,
SUBTYPES,
check_setting,
)
from brick_gl.infrastructure.repositories import utc
def _audit(repo, caller: dict, entity: str, key: str, before, after) -> None:
repo.audit_master(entity, key, before, after, who(caller), sub_of(caller), real_now())
def _clean(row: Optional[dict]) -> Optional[dict]:
return None if row is None else {k: v for k, v in row.items() if k not in ("created_at", "updated_at")}
def _dated(repo, valid_from: Optional[str], first: bool, what: str) -> date:
td = today(repo)
vf = dates.parse(valid_from) if valid_from else (td if first else td + dates.timedelta(days=1))
dates.check_dated(what, vf, td, first)
return vf
# ── companies and ledgers ────────────────────────────────────────────────────────────────────────────────────────────
def _company_view(repo, c: dict) -> dict:
ledgers = repo.company_ledgers.of_company(c["company_code"])
return {"company_code": c["company_code"], "name": c["name"], "country": c["country"],
"functional_currency": c["functional_currency"], "chart_id": c["chart_id"], "bp_id": c["bp_id"],
"financial_year_end_month": c["financial_year_end_month"], "default_profit_centre": c["default_profit_centre"],
"splitting_from_fiscal_year": c["splitting_from_fiscal_year"], "is_default": bool(c["is_default"]),
"ledgers": [r["ledger_id"] for r in ledgers]}
def _company_used(repo, code: str) -> bool:
from sqlalchemy import select
from brick_gl.infrastructure import tables as t
return repo.conn.execute(select(t.gl_journal.c.journal_id).where(t.gl_journal.c.company_code == code).limit(1)).first() is not None
@GL.use_case(kind="command", route="POST /companies", permission=COMPANY_WRITE, idempotent=True)
def keep_company(caller: dict, company_code: CompanyCode, name: Optional[text(200)] = None, country: Optional[text(2)] = None,
functional_currency: Optional[Currency] = None, chart_id: Optional[Code] = None,
financial_year_end_month: Optional[int] = None, bp_id: Optional[BigId] = None,
default_profit_centre: Optional[Code] = None, is_default: Optional[bool] = None) -> dict:
"""Add a company (a legal entity whose books are kept here, SAP's company code) or change one: its name, country,
functional currency, chart, financial year end month, the business partner that represents it (for intercompany) and
its default profit centre. The code is 1 to 4 upper-case letters or digits; the first company is the default company, to
which every posting that names none lands. The code and the functional currency are fixed once a journal uses them; the
year-end month is fixed once a year is opened. Fields left out keep their value."""
with repository() as repo, decide():
if not COMPANY_CODE.match(company_code):
raise RuleBroken("a company code is 1 to 4 upper-case letters or digits", status=422)
before = repo.companies.get(company_code)
row = dict(before) if before else {"company_code": company_code, "name": company_code, "country": None,
"functional_currency": None, "chart_id": None, "financial_year_end_month": 12,
"bp_id": None, "default_profit_centre": None, "splitting_from_fiscal_year": None,
"is_default": False, "active_until": None}
given = {"name": name, "country": country, "functional_currency": functional_currency, "chart_id": chart_id,
"financial_year_end_month": financial_year_end_month, "bp_id": bp_id, "default_profit_centre": default_profit_centre}
for k, v in given.items():
if v is not None:
row[k] = v
if not row["functional_currency"]:
row["functional_currency"] = repo.settings.in_force(today(repo)).functional_currency
if not CURRENCY.match(row["functional_currency"]):
raise RuleBroken(f"{row['functional_currency']!r} is not an ISO 4217 currency code", status=422)
if row["country"] is not None and not (len(row["country"]) == 2 and row["country"].isalpha() and row["country"].isupper()):
raise RuleBroken("country is an ISO 3166 code such as GB", status=422)
if not 1 <= int(row["financial_year_end_month"]) <= 12:
raise RuleBroken("financial_year_end_month is a month, 1 to 12", status=422)
if row["chart_id"] is None:
charts = [c for c in repo.charts.rows() if c["kind"] == "operational"]
if len(charts) != 1:
raise RuleBroken("name the company's chart (chart_id)", status=422)
row["chart_id"] = charts[0]["chart_id"]
chart = repo.charts.get(row["chart_id"])
if chart is None or chart["kind"] != "operational":
raise RuleBroken(f"{row['chart_id']} is not an operational chart: a company posts to an operational chart", status=422)
if before:
if _company_used(repo, company_code) and row["functional_currency"] != before["functional_currency"]:
raise RuleBroken(f"{company_code} has journals: its functional currency ({before['functional_currency']}) is fixed",
status=409)
if repo.years.of_company(company_code) and int(row["financial_year_end_month"]) != int(before["financial_year_end_month"]):
raise RuleBroken(f"{company_code} has opened a financial year: its year-end month is fixed", status=409)
if row["bp_id"] is not None:
other = [c for c in repo.companies.all() if c["bp_id"] == row["bp_id"] and c["company_code"] != company_code]
if other:
raise RuleBroken(f"bp_id {row['bp_id']} already represents company {other[0]['company_code']}", status=409)
existing = repo.companies.all()
make_default = is_default is True or (not existing and before is None)
if make_default:
for c in existing:
if c["is_default"] and c["company_code"] != company_code:
repo.companies.set((c["company_code"],), is_default=False)
row["is_default"] = True
elif is_default is False and before and before["is_default"]:
raise RuleBroken("exactly one company is the default: make another the default instead", status=409)
repo.companies.put({k: row[k] for k in ("company_code", "name", "country", "functional_currency", "chart_id",
"financial_year_end_month", "bp_id", "default_profit_centre",
"splitting_from_fiscal_year", "is_default", "active_until")})
# the leading ledger is mandatory: a bucket's first company brings one, and every company keeps it
lead = repo.ledgers.leading()
if lead is None:
repo.ledgers.put({"ledger_id": DEFAULT_LEDGER, "name": "Leading ledger", "kind": "leading",
"accounting_principle": "other", "reval_method": "closing_rate"})
lead = {"ledger_id": DEFAULT_LEDGER}
if repo.company_ledgers.get(company_code, lead["ledger_id"]) is None:
repo.company_ledgers.put({"company_code": company_code, "ledger_id": lead["ledger_id"], "from_fiscal_year": None,
"until_fiscal_year": None})
_audit(repo, caller, "company", company_code, _clean(before), _clean(row))
return _company_view(repo, repo.companies.get(company_code))
@GL.use_case(kind="query", route="GET /companies", permission=COMPANY_READ)
def list_companies(caller: dict) -> list:
"""The companies the caller may see: code, name, country, functional currency, chart, partner, year end and ledgers."""
with repository() as repo:
allowed = allowed_companies(repo, caller)
return [_company_view(repo, c) for c in repo.companies.all() if allowed is None or c["company_code"] in allowed]
@GL.use_case(kind="query", route="GET /companies/{company_code}", permission=COMPANY_READ)
def get_company(caller: dict, company_code: CompanyCode, on: Optional[IsoDate] = None) -> dict:
"""One company: its name, country, functional currency, chart and ledgers (what ar and payments read for the functional
currency), plus (v6) the statutory_tax_rate in force today (v7: on `on` when given, a date such as a period's last day; null when not
set) and the bucket's business_timezone. A company the caller may not see is a 404, not a 403."""
with repository() as repo:
c = company_row(repo, caller, company_code)
settings = repo.settings.in_force(dates.parse(on) or today(repo), c["company_code"])
return {**_company_view(repo, c), "statutory_tax_rate": settings.statutory_tax_rate, # v6: this route only, never the list
"business_timezone": settings.business_timezone}
@GL.use_case(kind="query", route="GET /companies/{company_code}/ledgers", permission=COMPANY_READ)
def get_company_ledgers(caller: dict, company_code: CompanyCode) -> dict:
"""The ledgers a company keeps and from which financial year (fixed-assets maps each depreciation area to one of them)."""
with repository() as repo:
c = company_row(repo, caller, company_code)
out = []
for r in repo.company_ledgers.of_company(company_code):
lg = repo.ledgers.get(r["ledger_id"])
out.append({"ledger_id": r["ledger_id"], "kind": lg["kind"], "accounting_principle": lg["accounting_principle"],
"from_fiscal_year": r["from_fiscal_year"], "until_fiscal_year": r["until_fiscal_year"]})
return {"company_code": company_code, "functional_currency": c["functional_currency"],
"fiscal_year_end_month": c["financial_year_end_month"], "ledgers": out}
@GL.use_case(kind="command", route="POST /ledgers", permission=COMPANY_WRITE, idempotent=True)
def keep_ledger(caller: dict, ledger_id: LedgerCode, name: Optional[text(200)] = None, kind: Optional[Code] = None,
accounting_principle: Optional[Code] = None, reval_method: Optional[Code] = None) -> dict:
"""Add a ledger (a set of books in one accounting principle: IFRS, LOCAL) or change its name, principle or revaluation
method (closing_rate, or losses_only as HGB section 256a requires). Exactly one ledger leads; its id is fixed once a
journal posts."""
with repository() as repo, decide():
if not LEDGER_CODE.match(ledger_id):
raise RuleBroken("a ledger id is 2 to 8 upper-case letters or digits", status=422)
before = repo.ledgers.get(ledger_id)
row = dict(before) if before else {"ledger_id": ledger_id, "name": ledger_id, "kind": "non_leading",
"accounting_principle": "other", "reval_method": "closing_rate"}
for k, v in (("name", name), ("kind", kind), ("accounting_principle", accounting_principle), ("reval_method", reval_method)):
if v is not None:
row[k] = v
if row["kind"] not in LEDGER_KINDS:
raise RuleBroken(f"kind is one of {', '.join(LEDGER_KINDS)}", status=422)
if row["reval_method"] not in REVAL_METHODS:
raise RuleBroken(f"reval_method is one of {', '.join(REVAL_METHODS)}", status=422)
lead = repo.ledgers.leading()
if row["kind"] == "leading" and lead and lead["ledger_id"] != ledger_id:
raise RuleBroken(f"exactly one ledger leads and it is {lead['ledger_id']}", status=409)
if before and before["kind"] != row["kind"]:
raise RuleBroken("a ledger's kind is fixed: the leading ledger stays leading", status=409)
repo.ledgers.put(row)
_audit(repo, caller, "ledger", ledger_id, _clean(before), _clean(row))
return _clean(repo.ledgers.get(ledger_id))
@GL.use_case(kind="command", route="POST /companies/{company_code}/ledgers", permission=COMPANY_WRITE, idempotent=True)
def assign_company_ledger(caller: dict, company_code: CompanyCode, ledger_id: LedgerCode,
from_fiscal_year: Optional[FiscalYearName] = None) -> dict:
"""Assign a ledger to a company from a financial year (FY2026). A ledger added later starts on the first day of a year,
opening with the leading ledger's carry-forward adjusted by ledger-specific journals dated that day. The leading ledger
is mandatory and cannot be removed."""
with repository() as repo, decide():
company_row(repo, caller, company_code)
lg = repo.ledgers.get(ledger_id)
if lg is None:
raise Refused(404, f"no ledger {ledger_id}")
before = repo.company_ledgers.get(company_code, ledger_id)
if before and before["from_fiscal_year"] != from_fiscal_year and lg["kind"] == "leading":
raise RuleBroken("the leading ledger is mandatory from the start and cannot be moved", status=409)
if before:
return _clean(before)
if lg["kind"] != "leading" and from_fiscal_year is None:
raise RuleBroken("name the financial year the ledger starts in (from_fiscal_year, FY2026)", status=422)
if from_fiscal_year:
started = [y for y in repo.years.of_company(company_code) if y["status"] == "closed"]
if any(dates.fy_number(y["fiscal_year"]) >= dates.fy_number(from_fiscal_year) for y in started):
raise RuleBroken("a ledger starts in a year that is not yet closed", status=409)
row = {"company_code": company_code, "ledger_id": ledger_id, "from_fiscal_year": from_fiscal_year, "until_fiscal_year": None}
repo.company_ledgers.put(row)
_audit(repo, caller, "company_ledger", f"{company_code}:{ledger_id}", None, row)
return row
# ── charts and accounts ──────────────────────────────────────────────────────────────────────────────────────────────
@GL.use_case(kind="command", route="POST /charts", permission=ACCOUNT_WRITE, idempotent=True)
def keep_chart(caller: dict, chart_id: Code, name: Optional[text(200)] = None, kind: Optional[Code] = None,
country: Optional[text(2)] = None) -> dict:
"""Add a chart of accounts - operational (companies post to it, and share it if they like) or group (consolidation
reports in it; no line is ever posted to it) - or change its name. The kind is fixed."""
with repository() as repo, decide():
before = repo.charts.get(chart_id)
row = dict(before) if before else {"chart_id": chart_id, "name": chart_id, "kind": "operational", "country": None}
for k, v in (("name", name), ("kind", kind), ("country", country)):
if v is not None:
row[k] = v
if row["kind"] not in CHART_KINDS:
raise RuleBroken(f"kind is one of {', '.join(CHART_KINDS)}", status=422)
if before and before["kind"] != row["kind"]:
raise RuleBroken("a chart's kind is fixed", status=409)
repo.charts.put({k: row[k] for k in ("chart_id", "name", "kind", "country")})
_audit(repo, caller, "chart", chart_id, _clean(before), _clean(row))
return _clean(repo.charts.get(chart_id))
def first_operational_chart(repo) -> Optional[str]:
charts = sorted((c for c in repo.charts.rows() if c["kind"] == "operational"), key=lambda c: (c["created_at"] or 0, c["chart_id"]))
return charts[0]["chart_id"] if charts else None
def account_id_for(repo, chart_id: str, code: str) -> int:
"""derived_id("account", code) in the first operational chart (so v2's ids stand), derived_id("account", chart_id, code) in any
other: a code names one account for ever."""
return derived_id("account", code) if first_operational_chart(repo) == chart_id else derived_id("account", chart_id, code)
def _account_view(repo, a: dict) -> dict:
out = {k: a[k] for k in ("account_id", "chart_id", "code", "name", "type", "subtype", "normal_balance", "currency",
"parent_account_id", "monetary", "source_only", "fs_cost_centre", "fs_profit_centre", "fs_bp_id",
"fs_tax_code", "default_profit_centre", "active_until")}
return out
@GL.use_case(kind="command", route="POST /accounts", permission=ACCOUNT_WRITE, idempotent=True)
def keep_account(caller: dict, chart_id: Code, code: Code, name: Optional[text(200)] = None, type: Optional[Code] = None,
subtype: Optional[Code] = None, parent_code: Optional[Code] = None, currency: Optional[Currency] = None,
monetary: Optional[bool] = None, source_only: Optional[bool] = None, cost_centre: Optional[Code] = None,
profit_centre: Optional[Code] = None, partner: Optional[Code] = None, tax_code: Optional[Code] = None,
default_profit_centre: Optional[Code] = None, active_until: Optional[IsoDate] = None) -> dict:
"""Add an account to a chart, or change one: the customer's own code, a type (asset, liability, equity, income, expense),
a subtype (receivable, payable, bank, tax, retained_earnings, asset_cost, accumulated_depreciation, ic, other), a place in
the tree (parent_code; the edge is written with the account), whether it is revalued (monetary) or kept by its module
(source_only), and its field status for cost centre, profit centre, partner and tax code (required, optional, forbidden).
The account keeps a stable account_id for ever. An account a journal has used cannot change type or code - only its name,
or be deactivated from a day after today (active_until)."""
with repository() as repo, decide():
chart = repo.charts.get(chart_id)
if chart is None:
raise Refused(404, f"no chart {chart_id}")
before = repo.accounts.by_code(chart_id, code)
row = dict(before) if before else {
"account_id": account_id_for(repo, chart_id, code), "chart_id": chart_id, "code": code, "name": code, "type": "asset",
"subtype": "other", "normal_balance": None, "currency": None, "parent_account_id": None, "monetary": False,
"source_only": False, "fs_cost_centre": "optional", "fs_profit_centre": "optional", "fs_bp_id": "optional",
"fs_tax_code": "optional", "default_profit_centre": None, "movement_tracked": False, "active_until": None}
if before is None and type is None:
raise RuleBroken("name the account's type (asset, liability, equity, income, expense)", status=422)
given = {"name": name, "type": type, "subtype": subtype, "currency": currency, "monetary": monetary,
"source_only": source_only, "fs_cost_centre": cost_centre, "fs_profit_centre": profit_centre,
"fs_bp_id": partner, "fs_tax_code": tax_code, "default_profit_centre": default_profit_centre}
for k, v in given.items():
if v is not None:
row[k] = v
if row["type"] not in ACCOUNT_TYPES:
raise RuleBroken(f"type is one of {', '.join(ACCOUNT_TYPES)}", status=422)
if row["subtype"] not in SUBTYPES:
raise RuleBroken(f"subtype is one of {', '.join(SUBTYPES)}", status=422)
for f in ("fs_cost_centre", "fs_profit_centre", "fs_bp_id", "fs_tax_code"):
if row[f] not in FIELD_STATUS:
raise RuleBroken(f"a field status is one of {', '.join(FIELD_STATUS)}", status=422)
if row["currency"] and not CURRENCY.match(row["currency"]):
raise RuleBroken(f"{row['currency']!r} is not an ISO 4217 currency code", status=422)
row["normal_balance"] = "debit" if row["type"] in ("asset", "expense") else "credit"
if row["subtype"] in ("receivable", "payable", "asset_cost", "accumulated_depreciation"):
row["source_only"] = True if source_only is None else row["source_only"]
if before and repo.accounts.used(before["account_id"]) and (row["type"] != before["type"]):
raise RuleBroken(f"account {code} is used by a journal: its type is fixed (change its name, or deactivate it)",
status=409)
if active_until:
row["active_until"] = _dated(repo, active_until, False, "a deactivation")
if parent_code:
parent = repo.accounts.by_code(chart_id, parent_code)
if parent is None:
raise Refused(404, f"no account {parent_code} in chart {chart_id}")
if parent["account_id"] == row["account_id"]:
raise RuleBroken("an account cannot be its own parent", status=422)
anc = parent
while anc is not None and anc["parent_account_id"] is not None:
if anc["parent_account_id"] == row["account_id"]:
raise RuleBroken("the tree would loop: that parent is a descendant of this account", status=422)
anc = repo.accounts.get(anc["parent_account_id"])
row["parent_account_id"] = parent["account_id"]
repo.accounts.put(row)
if parent_code and row["parent_account_id"] is not None:
repo.accounts.set_edge(row["parent_account_id"], row["account_id"]) # the tree, in the same use case
_audit(repo, caller, "account", f"{chart_id}:{code}", _clean(before), _clean(row))
return _account_view(repo, row)
@GL.use_case(kind="command", route="POST /accounts/change-code", permission=ACCOUNT_WRITE)
def rename_account_code(caller: dict, chart_id: Code, code: Code, new_code: Code) -> dict:
"""Change the code of an account no journal has used. A used account keeps its code for ever (its account_id and every
line stay readable under the number the books were kept in)."""
with repository() as repo, decide():
a = repo.accounts.by_code(chart_id, code)
if a is None:
raise Refused(404, f"no account {code} in chart {chart_id}")
if repo.accounts.used(a["account_id"]):
raise RuleBroken(f"account {code} is used by a journal: its code is fixed", status=409)
if repo.accounts.by_code(chart_id, new_code):
raise RuleBroken(f"chart {chart_id} already has an account {new_code}", status=409)
repo.accounts.set((a["account_id"],), code=new_code)
_audit(repo, caller, "account", f"{chart_id}:{code}", {"code": code}, {"code": new_code})
return _account_view(repo, repo.accounts.get(a["account_id"]))
@GL.use_case(kind="query", route="GET /charts/{chart_id}/accounts", permission=COMPANY_READ)
def list_chart_accounts(caller: dict, chart_id: Code) -> list:
"""The accounts of a chart in code order (the group chart's accounts are what consolidation reports in)."""
with repository() as repo:
if repo.charts.get(chart_id) is None:
raise Refused(404, f"no chart {chart_id}")
return [{"account_id": a["account_id"], "code": a["code"], "name": a["name"], "type": a["type"], "subtype": a["subtype"]}
for a in repo.accounts.of_chart(chart_id)]
@GL.use_case(kind="query", route="GET /accounts/{account_id}", permission=COMPANY_READ)
def get_account(caller: dict, account_id: BigId) -> dict:
"""One account with its place in the tree, its group account and alternative numbers (with history)."""
with repository() as repo:
a = repo.accounts.get(int(account_id))
if a is None:
raise Refused(404, f"no account {account_id}")
out = _account_view(repo, a)
out["children"] = repo.accounts.children(a["account_id"])
out["group_account_history"] = [{"valid_from": r["valid_from"], "group_account_id": r["group_account_id"]}
for r in repo.group_map.history(a["account_id"])]
out["alt_numbers"] = [{"company_code": r["company_code"], "valid_from": r["valid_from"], "alt_code": r["alt_code"],
"alt_name": r["alt_name"]} for c in visible_codes(repo, caller)
for r in repo.alt_accounts.history(c, a["account_id"])]
return out
@GL.use_case(kind="command", route="POST /accounts/group-mapping", permission=ACCOUNT_WRITE)
def map_group_account(caller: dict, account_id: BigId, group_account_id: BigId, valid_from: Optional[IsoDate] = None) -> dict:
"""Map an operational account to the group account it reports under, dated from a day after today (the first row for an
account may start today); the history is readable. Consolidation reads the mapping through the group extract: every
account with a posting and no group account is listed as unmapped."""
with repository() as repo, decide():
a, g = repo.accounts.get(int(account_id)), repo.accounts.get(int(group_account_id))
if a is None or g is None:
raise Refused(404, "no such account")
if repo.charts.get(g["chart_id"])["kind"] != "group":
raise RuleBroken(f"{g['code']} is not an account of a group chart", status=422)
if repo.charts.get(a["chart_id"])["kind"] == "group":
raise RuleBroken(f"{a['code']} is a group account: map an operational account", status=422)
first = not repo.group_map.history(a["account_id"])
vf = _dated(repo, valid_from, first, "a group mapping")
row = {"account_id": a["account_id"], "valid_from": vf, "group_account_id": g["account_id"], "set_by": who(caller),
"set_at": utc(real_now())}
if repo.group_map.get(a["account_id"], vf):
raise RuleBroken(f"{a['code']} already has a group mapping from {vf}", status=409)
repo.group_map.add(row)
_audit(repo, caller, "group_account_map", f"{a['account_id']}:{vf}", None, {"group_account_id": g["account_id"]})
return {"account_id": a["account_id"], "valid_from": vf, "group_account_id": g["account_id"]}
@GL.use_case(kind="command", route="POST /accounts/alternative-number", permission=ACCOUNT_WRITE)
def set_alt_account(caller: dict, company_code: CompanyCode, account_id: BigId, alt_code: Code, alt_name: Optional[text(200)] = None,
valid_from: Optional[IsoDate] = None) -> dict:
"""Give an account the alternative (country) number a company's statutory files use (DE01's SKR04 4400 for 4000), dated
from a day after today; the history is readable. The FEC and SAF-T exports of the company use it."""
with repository() as repo, decide():
company_row(repo, caller, company_code)
a = repo.accounts.get(int(account_id))
if a is None:
raise Refused(404, f"no account {account_id}")
first = not repo.alt_accounts.history(company_code, a["account_id"])
vf = _dated(repo, valid_from, first, "an alternative number")
if repo.alt_accounts.get(company_code, a["account_id"], vf):
raise RuleBroken(f"{a['code']} already has an alternative number for {company_code} from {vf}", status=409)
row = {"company_code": company_code, "account_id": a["account_id"], "valid_from": vf, "alt_code": alt_code,
"alt_name": alt_name or a["name"]}
repo.alt_accounts.add(row)
_audit(repo, caller, "alt_account", f"{company_code}:{a['account_id']}:{vf}", None, row)
return row
# ── the account map (GL-12) ──────────────────────────────────────────────────────────────────────────────────────────
@GL.use_case(kind="command", route="POST /account-map", permission=SETTINGS_WRITE)
def set_gl_account_map(caller: dict, chart_id: Code, purpose: Code, account_id: BigId, qualifier: Optional[Code] = None,
valid_from: Optional[IsoDate] = None) -> dict:
"""Map a posting purpose (receivable, revenue, tax + tax_code, bank + bank account, bad_debt, fx_gain, fx_loss, rounding,
retained_earnings, split_clearing, ic_receivable / ic_payable + partner company, fixed-assets' purposes + asset class ...)
to an account of the chart, dated from a day after today (the first row of a purpose may start today). Postings dated on
or after that day resolve the purpose to that account; earlier postings are untouched. The map is per chart, so companies
sharing a chart share it."""
with repository() as repo, decide():
chart = repo.charts.get(chart_id)
if chart is None:
raise Refused(404, f"no chart {chart_id}")
if chart["kind"] == "group":
raise RuleBroken("a group chart takes no postings, so it has no account map", status=422)
a = repo.accounts.get(int(account_id))
if a is None or a["chart_id"] != chart_id:
raise RuleBroken(f"account {account_id} is not an account of chart {chart_id}", status=422)
q = qualifier or ""
first = not repo.account_map.has_any(chart_id, purpose, q)
vf = _dated(repo, valid_from, first, "an account-map row")
if repo.account_map.get(chart_id, purpose, q, vf):
raise RuleBroken("the account map already has a row for that purpose and day", status=409)
row = {"chart_id": chart_id, "purpose": purpose, "qualifier": q, "valid_from": vf, "account_id": a["account_id"],
"set_by": who(caller), "set_at": utc(real_now())}
repo.account_map.add(row)
_audit(repo, caller, "account_map", f"{chart_id}:{purpose}:{q}:{vf}", None, {"account_id": a["account_id"]})
return {"chart_id": chart_id, "purpose": purpose, "qualifier": q, "valid_from": vf, "account_id": a["account_id"]}
@GL.use_case(kind="query", route="GET /account-map", permission=COMPANY_READ)
def get_gl_account_map(caller: dict, chart_id: Code) -> dict:
"""Where each purpose posts in a chart, every row with its valid_from, and the PURPOSES the composed modules post that have
no row yet (unmapped): readiness shows them before the first posting."""
with repository() as repo:
if repo.charts.get(chart_id) is None:
raise Refused(404, f"no chart {chart_id}")
rows = repo.account_map.rows_of(chart_id)
mapped = {r["purpose"] for r in rows}
return {"chart_id": chart_id,
"rows": [{k: r[k] for k in ("purpose", "qualifier", "valid_from", "account_id", "set_by")} for r in rows],
"unmapped": [p for p in PURPOSES if p not in mapped]}
# ── dimensions (GL-43) ───────────────────────────────────────────────────────────────────────────────────────────────
@GL.use_case(kind="command", route="POST /segments", permission=ACCOUNT_WRITE, idempotent=True)
def keep_segment(caller: dict, segment: Code, name: Optional[text(200)] = None, active_until: Optional[IsoDate] = None) -> dict:
"""Add a reportable segment (IFRS 8) or change its name."""
with repository() as repo, decide():
before = repo.dims.segments.get(segment)
row = {"segment": segment, "name": name or (before or {}).get("name") or segment,
"active_until": dates.parse(active_until) if active_until else (before or {}).get("active_until")}
repo.dims.segments.put(row)
_audit(repo, caller, "segment", segment, _clean(before), row)
return row
@GL.use_case(kind="command", route="POST /profit-centres", permission=ACCOUNT_WRITE, idempotent=True)
def keep_profit_centre(caller: dict, profit_centre: Code, name: Optional[text(200)] = None, parent_profit_centre: Optional[Code] = None,
company_codes: Optional[text(200)] = None, valid_from: Optional[IsoDate] = None,
valid_until: Optional[IsoDate] = None) -> dict:
"""Add a profit centre (an area of responsibility whose result, and with splitting its balance sheet, is reported) or change
it: its name, parent, the companies it is valid in (comma-separated; none = all) and its validity dates. A profit centre
not valid for the company on the posting date refuses the line (gate profit_centre_valid)."""
with repository() as repo, decide():
before = repo.dims.profit_centres.get(profit_centre)
row = dict(before) if before else {"profit_centre": profit_centre, "name": profit_centre, "parent_profit_centre": None,
"company_codes": None, "valid_from": None, "valid_until": None}
if parent_profit_centre and repo.dims.profit_centres.get(parent_profit_centre) is None:
raise Refused(404, f"no profit centre {parent_profit_centre}")
for k, v in (("name", name), ("parent_profit_centre", parent_profit_centre), ("company_codes", company_codes)):
if v is not None:
row[k] = v
if valid_from:
row["valid_from"] = dates.parse(valid_from)
if valid_until:
row["valid_until"] = dates.parse(valid_until)
repo.dims.profit_centres.put({k: row[k] for k in ("profit_centre", "name", "parent_profit_centre", "company_codes", "valid_from", "valid_until")})
_audit(repo, caller, "profit_centre", profit_centre, _clean(before), _clean(row))
return {k: row[k] for k in ("profit_centre", "name", "parent_profit_centre", "company_codes", "valid_from", "valid_until")}
@GL.use_case(kind="command", route="POST /profit-centres/{profit_centre}/segment", permission=ACCOUNT_WRITE)
def assign_profit_centre_segment(caller: dict, profit_centre: Code, segment: Code, valid_from: Optional[IsoDate] = None) -> dict:
"""Say which segment a profit centre belongs to, dated: re-assigning one never moves posted lines (the first assignment may
start today, a later one from a day after today)."""
with repository() as repo, decide():
if repo.dims.profit_centres.get(profit_centre) is None:
raise Refused(404, f"no profit centre {profit_centre}")
if repo.dims.segments.get(segment) is None:
raise Refused(404, f"no segment {segment}")
first = not repo.dims.pc_segments.rows(repo.dims.pc_segments.table.c.profit_centre == profit_centre)
vf = _dated(repo, valid_from, first, "a segment assignment")
if repo.dims.pc_segments.get(profit_centre, vf):
raise RuleBroken(f"{profit_centre} already has a segment from {vf}", status=409)
repo.dims.pc_segments.add({"profit_centre": profit_centre, "valid_from": vf, "segment": segment})
_audit(repo, caller, "profit_centre_segment", f"{profit_centre}:{vf}", None, {"segment": segment})
return {"profit_centre": profit_centre, "segment": segment, "valid_from": vf}
@GL.use_case(kind="command", route="POST /cost-centres", permission=ACCOUNT_WRITE, idempotent=True)
def keep_cost_centre(caller: dict, cost_centre: Code, name: Optional[text(200)] = None, company_code: Optional[CompanyCode] = None,
active_until: Optional[IsoDate] = None) -> dict:
"""Add a cost centre (a department, site or project a line can be reported by) or change its name, company or end date.
hr, purchasing, fixed-assets and inventory store the code and ask get_cost_centre whether it is open."""
with repository() as repo, decide():
before = repo.dims.cost_centres.get(cost_centre)
row = dict(before) if before else {"cost_centre": cost_centre, "name": cost_centre, "company_code": None, "active_until": None}
if company_code is not None:
if repo.companies.get(company_code) is None:
raise Refused(404, f"no company {company_code}")
row["company_code"] = company_code
if name is not None:
row["name"] = name
if active_until:
row["active_until"] = dates.parse(active_until)
repo.dims.cost_centres.put({k: row[k] for k in ("cost_centre", "name", "company_code", "active_until")})
_audit(repo, caller, "cost_centre", cost_centre, _clean(before), _clean(row))
return {k: row[k] for k in ("cost_centre", "name", "company_code", "active_until")}
@GL.use_case(kind="command", route="POST /cost-centres/{cost_centre}/profit-centre", permission=ACCOUNT_WRITE)
def assign_cost_centre(caller: dict, cost_centre: Code, profit_centre: Code, valid_from: Optional[IsoDate] = None) -> dict:
"""Say which profit centre a cost centre belongs to, dated: a line with this cost centre and no profit centre gets that
profit centre (and its segment) from the assignment in force on the journal date."""
with repository() as repo, decide():
if repo.dims.cost_centres.get(cost_centre) is None:
raise Refused(404, f"no cost centre {cost_centre}")
if repo.dims.profit_centres.get(profit_centre) is None:
raise Refused(404, f"no profit centre {profit_centre}")
first = not repo.dims.cc_assignments.rows(repo.dims.cc_assignments.table.c.cost_centre == cost_centre)
vf = _dated(repo, valid_from, first, "a cost-centre assignment")
if repo.dims.cc_assignments.get(cost_centre, vf):
raise RuleBroken(f"{cost_centre} already has an assignment from {vf}", status=409)
repo.dims.cc_assignments.add({"cost_centre": cost_centre, "valid_from": vf, "profit_centre": profit_centre})
_audit(repo, caller, "cost_centre_assignment", f"{cost_centre}:{vf}", None, {"profit_centre": profit_centre})
return {"cost_centre": cost_centre, "profit_centre": profit_centre, "valid_from": vf}
@GL.use_case(kind="query", route="GET /cost-centres/{cost_centre}", permission=COST_CENTRE_READ)
def get_cost_centre(caller: dict, cost_centre: Code) -> dict:
"""Whether a cost centre exists and is open (inventory's and purchasing's gate before a goods issue): status open or
closed with its company; a closed one is answered, never a 404. gl writes nothing."""
with repository() as repo:
r = repo.dims.cost_centres.get(cost_centre)
if r is None:
raise Refused(404, f"no cost centre {cost_centre}")
closed = r["active_until"] is not None and r["active_until"] <= today(repo)
return {"cost_centre": cost_centre, "name": r["name"], "status": "closed" if closed else "open", "company_code": r["company_code"]}
# ── settings and the owner's controls (GL-14) ────────────────────────────────────────────────────────────────────────
def _set(repo, caller: dict, key: str, value, valid_from: Optional[str], scope: str, control: bool) -> dict:
if key not in SETTINGS:
raise RuleBroken(f"there is no setting {key}", status=422)
if SETTINGS[key][2] != control:
if control:
raise RuleBroken(f"{key} is a day-to-day setting: change it with set_gl_setting", status=422)
raise RuleBroken(f"{key} is a control on the people who keep the books: only the owner (gl:settings:controls) changes it "
"- see set_gl_control", status=403)
try:
value = check_setting(key, value, scope)
except ValueError as e:
raise RuleBroken(str(e), status=422) from None
if scope and repo.companies.get(scope) is None:
raise Refused(404, f"no company {scope}")
first = not repo.settings.has_any(key, scope)
td = today(repo)
vf = dates.parse(valid_from) if valid_from else (td if first else td + dates.timedelta(days=1))
dates.check_dated(f"a {key} row", vf, td, first)
if key == "document_splitting":
_check_splitting_start(repo, scope, vf, value)
if key == "functional_currency" and repo.companies.all() and any(_company_used(repo, c["company_code"]) for c in repo.companies.all()):
raise RuleBroken("the functional currency is fixed once a journal uses it (set it per company)", status=409)
if repo.settings.get(key, scope, vf):
raise RuleBroken(f"{key} already has a row from {vf}", status=409)
repo.settings.add_row(key, scope, vf, value, who(caller), real_now())
_audit(repo, caller, "setting", f"{key}:{scope}:{vf}", None, {"value": value})
return {"setting_key": key, "scope": scope, "valid_from": vf, "value": value}
def _check_splitting_start(repo, scope: str, vf: date, value) -> None:
"""Document splitting is switched on only from a fiscal year's first day, and never off within a year (GL-22)."""
codes = [scope] if scope else [c["company_code"] for c in repo.companies.all()]
for code in codes:
c = repo.companies.get(code)
start, _ = dates.fy_bounds(dates.fiscal_year_of(vf, int(c["financial_year_end_month"])), int(c["financial_year_end_month"]))
if vf != start:
raise RuleBroken(f"document splitting changes only on the first day of a financial year: {code}'s year starts on "
f"{start}, not {vf}", status=422)
if value == "on" and scope:
c = repo.companies.get(scope)
repo.companies.set((scope,), splitting_from_fiscal_year=dates.fy_name(
dates.fiscal_year_of(vf, int(c["financial_year_end_month"]))))
@GL.use_case(kind="command", route="POST /settings", permission=SETTINGS_WRITE)
def set_gl_setting(caller: dict, key: text(80), value: Optional[bool | int | str], valid_from: Optional[IsoDate] = None,
company_code: Optional[CompanyCode] = None) -> dict:
"""Change a day-to-day setting (ic_tolerance, show_result_in_equity, revaluation) as a new dated row, from
tomorrow by default (the first row of a key may start today; a row dated today or earlier is refused). The owner's
controls are refused here: they are set with set_gl_control by whoever holds gl:settings:controls."""
with repository() as repo, decide():
return _set(repo, caller, key, value, valid_from, company_code or "", control=False)
@GL.use_case(kind="command", route="POST /controls", permission=CONTROLS)
def set_gl_control(caller: dict, key: text(80), value: Optional[bool | int | str], valid_from: Optional[IsoDate] = None,
company_code: Optional[CompanyCode] = None) -> dict:
"""The owner changes a CONTROL on the people who keep the books (manual_journal_approval, manual_journal_threshold,
reopen_second_person, gap_free_numbers, allocation_cycle_approval, soft_close, document_splitting, close_cockpit, ic_match_required,
business_timezone, financial_year_end_month, functional_currency, the AI step modes) as a new dated row. A control dated
today or earlier is refused, the first row of a key excepted: it takes effect from tomorrow, never today."""
with repository() as repo, decide():
return _set(repo, caller, key, value, valid_from, company_code or "", control=True)
@GL.use_case(kind="query", route="GET /settings", permission=COMPANY_READ)
def list_gl_settings(caller: dict, key: Optional[text(80)] = None, company_code: Optional[CompanyCode] = None) -> dict:
"""Every setting row with its valid_from and who set it (the history of a control is readable), and the value in force
today for each key."""
with repository() as repo:
td = today(repo)
scope = company_code
rows = repo.settings.history(key, None)
if scope:
rows = [r for r in rows if r["scope"] in ("", scope)]
in_force = repo.settings.in_force(td, scope or "")
return {"today": td, "in_force": {k: in_force.get(k) for k in SETTINGS if not key or k == key},
"controls": sorted(CONTROL_KEYS),
"history": [{k: r[k] for k in ("setting_key", "scope", "valid_from", "value", "set_by")} for r in rows]}
@GL.use_case(kind="command", route="POST /company-access", permission=CONTROLS)
def grant_company_access(caller: dict, principal_sub: text(200, min_length=1), company_code: CompanyCode,
principal_email: Optional[text(254)] = None, valid_from: Optional[IsoDate] = None,
valid_until: Optional[IsoDate] = None) -> dict:
"""The owner says which companies a person may see and post in, dated from tomorrow (the first row may start today). From
then on every read and write outside those companies answers 404 for that person, the trial balance and reports list
their companies only, and a cross-company journal touching another company is refused. A person with no row sees every
company (the default, for single-company books). Held under gl:settings:controls, never by whoever it controls."""
with repository() as repo, decide():
if repo.companies.get(company_code) is None:
raise Refused(404, f"no company {company_code}")
first = not [r for r in repo.access.rows() if r["principal_sub"] == principal_sub]
vf = _dated(repo, valid_from, first, "company access")
if repo.access.get(principal_sub, company_code, vf):
raise RuleBroken("that person already has a row for the company from that day", status=409)
until = dates.parse(valid_until) if valid_until else None
row = {"principal_sub": principal_sub, "company_code": company_code, "valid_from": vf, "principal_email": principal_email,
"valid_until": until, "set_by": who(caller), "set_at": utc(real_now())}
repo.access.add(row)
_audit(repo, caller, "company_access", f"{principal_sub}:{company_code}:{vf}", None, row)
return {k: row[k] for k in ("principal_sub", "company_code", "valid_from", "principal_email", "valid_until")}
@GL.use_case(kind="query", route="GET /company-access", permission=CONTROLS)
def list_company_access(caller: dict) -> dict:
"""Every company-access row with its dates (the owner's view)."""
with repository() as repo:
return {"rows": [{k: r[k] for k in ("principal_sub", "principal_email", "company_code", "valid_from", "valid_until", "set_by")}
for r in repo.access.rows()]}
# ── FX rates (GL-33) ─────────────────────────────────────────────────────────────────────────────────────────────────
@GL.use_case(kind="command", route="POST /fx-rates", permission=FX_WRITE)
def set_fx_rate(caller: dict, from_currency: Currency, rate: FxRate, rate_date: IsoDate, to_currency: Optional[Currency] = None,
rate_type: Optional[Code] = None, source: Optional[text(100)] = None) -> dict:
"""Enter the rate that restates a currency on a date: units of `to_currency` (default: the default company's functional
currency) for one `from_currency`, by type - spot (every posting), average (one per period, dated its last day) or closing
(dated the period end; spot on that date when absent). A rate is a fact dated by its own date: today or earlier, never
later, and never changed - a posting, a revaluation or a translation may have used it."""
with repository() as repo, decide():
td = today(repo)
d = dates.parse(rate_date)
if d > td:
raise RuleBroken(f"a rate is dated by its own date, today ({td}) or earlier, not {d}", status=422)
if to_currency is None:
default = repo.companies.default()
to_currency = default["functional_currency"] if default else repo.settings.in_force(td).functional_currency
rt = rate_type or "spot"
if rt not in RATE_TYPES:
raise RuleBroken(f"rate_type is one of {', '.join(RATE_TYPES)}", status=422)
if from_currency == to_currency or not CURRENCY.match(from_currency) or not CURRENCY.match(to_currency):
raise RuleBroken("a rate restates one currency in another: two different ISO 4217 codes", status=422)
if repo.fx.get(from_currency, to_currency, rt, d):
raise RuleBroken(f"there is already a {rt} rate for {from_currency} to {to_currency} on {d}: a rate is never changed "
"(a posting or revaluation may have used it)", status=409)
row = {"from_ccy": from_currency, "to_ccy": to_currency, "rate_type": rt, "rate_date": d, "rate": rate,
"source": source or "manual", "entered_by": who(caller), "entered_sub": sub_of(caller), "entered_at": utc(real_now())}
repo.fx.add(row)
_audit(repo, caller, "fx_rate", f"{from_currency}:{to_currency}:{rt}:{d}", None, {"rate": str(rate)})
return {"currency": from_currency, "to": to_currency, "type": rt, "rate_date": d, "rate": rate, "source": row["source"]}
@GL.use_case(kind="query", route="GET /fx-rates/{currency}", permission=FX_READ)
def get_fx_rate(caller: dict, currency: Currency, on: Optional[IsoDate] = None, type: Optional[Code] = None,
to: Optional[Currency] = None, month: Optional[text(7)] = None) -> dict:
"""The rate in force for a currency on a date - what tax and consolidation read so that a document and its journal use the
same rate: {currency, to, type, on, rate_date, rate, source}. `type` is spot (default), average or closing (a closing
rate falls back to spot on that date); `to` defaults to the default company's functional currency. 404 with the fix when
no rate of that type is dated on or before the date; rate 1 when the currency equals `to`. `on` answers the NEWEST rate of the type dated
on or before it (rate_date says which), never a later one. v6: for type=average, `month` (YYYY-MM, instead of `on`) answers exactly the
average dated that month's last day, and 404s when there is none: never the nearest."""
with repository() as repo:
rt = type or "spot"
if rt not in RATE_TYPES:
raise Refused(422, f"type is one of {', '.join(RATE_TYPES)}")
if on and month:
raise Refused(422, "send on OR month, not both")
if month and rt != "average":
raise Refused(422, "month addresses a monthly average: send type=average with it")
if month:
if not re.fullmatch(r"[0-9]{4}-(0[1-9]|1[0-2])", month):
raise Refused(422, "month is YYYY-MM")
first = date(int(month[:4]), int(month[5:]), 1)
d = (first.replace(year=first.year + (first.month == 12), month=first.month % 12 + 1)) - dates.timedelta(days=1)
else:
d = dates.parse(on) if on else today(repo)
default = repo.companies.default()
target = to or (default["functional_currency"] if default else repo.settings.in_force(d).functional_currency)
if currency == target:
return {"currency": currency, "to": target, "type": rt, "on": d, "rate_date": d, "rate": "1", "source": "identity"}
found = repo.fx.get(currency, target, rt, d) if month else repo.fx.in_force(currency, target, rt, d)
if found is None and month:
raise Refused(404, f"[fx_rate] no average rate for {currency} to {target} is dated {d}, the last day of {month}",
gate="fx_rate", fix="Currencies: enter the month's average rate; the nearest one is never used")
if found is None and rt == "closing":
found = repo.fx.in_force(currency, target, "spot", d)
if found is None:
raise Refused(404, f"[fx_rate] no {rt} rate for {currency} to {target} is dated on or before {d}", gate="fx_rate",
fix="Currencies: enter the rate for the date")
return {"currency": currency, "to": target, "type": rt, "on": d, "rate_date": found["rate_date"], "rate": found["rate"],
"source": found["source"]}
@GL.use_case(kind="query", route="GET /fx-rates", permission=FX_READ)
def list_fx_rates(caller: dict, currency: Optional[Currency] = None, to: Optional[Currency] = None) -> dict:
"""The rates entered, oldest first, optionally for one currency."""
with repository() as repo:
return {"rates": [{"currency": r["from_ccy"], "to": r["to_ccy"], "type": r["rate_type"], "rate_date": r["rate_date"],
"rate": r["rate"], "source": r["source"]} for r in repo.fx.listing(currency, to)]}