dodil/erp-brick-gl
DODIL ERP brick: general ledger
src/brick_gl/infrastructure/repositories.py
"""The repositories: what the use cases read and write, on one connection (what `use_cases.repository()` yields).
A repository never decides a business rule; it finds rows and writes them. Facts are INSERTED (a conflict on the key is the
replay or the race lost); configuration is upserted with every value in VALUES (DataK3 writes every column named there);
a status that moves is an UPDATE by primary key. There is no read-your-writes inside an open transaction, so a use case that
writes rows then reports them works the answer out instead of querying it back.
"""
import json
import uuid
from datetime import date, datetime, timezone
from decimal import Decimal
from typing import Optional
from erp_platform.repository import upsert
from sqlalchemy import and_, func, insert, or_, select, type_coerce, update
from brick_gl.domain import dates
from brick_gl.domain.money import ZERO, q2
from brick_gl.domain.values import SETTINGS, Settings
from brick_gl.infrastructure import tables as t
from brick_gl.infrastructure.tables import MONEY
GENESIS = "0" * 64
def utc(v: Optional[datetime]) -> Optional[datetime]:
"""Stored timestamps are naive UTC."""
if v is None:
return None
return v.astimezone(timezone.utc).replace(tzinfo=None) if v.tzinfo else v
def _stamp() -> dict:
now = datetime.now(timezone.utc).replace(tzinfo=None)
return {"created_at": now, "updated_at": now}
def _d(row) -> Optional[dict]:
return dict(row._mapping) if row is not None else None
def _money(v) -> Decimal:
return q2(v) if v is not None else ZERO
class Store:
"""Generic keyed access over one table."""
table = None
key: tuple = ()
def __init__(self, conn):
self.conn = conn
def get(self, *key) -> Optional[dict]:
cond = and_(*(self.table.c[k] == v for k, v in zip(self.key, key)))
return _d(self.conn.execute(select(self.table).where(cond)).first())
def rows(self, *conds, order=None, limit: Optional[int] = None) -> list:
q = select(self.table).where(and_(*conds)) if conds else select(self.table)
for o in (order or [self.table.c[k] for k in self.key]):
q = q.order_by(o)
if limit:
q = q.limit(limit)
return [dict(r._mapping) for r in self.conn.execute(q)]
def add(self, row: dict) -> None:
"""A plain INSERT: the primary key is the guard; a second writer of the key fails (23505)."""
self.conn.execute(insert(self.table).values(**{k: v for k, v in {**row, **_stamp()}.items()}))
def add_many(self, rows: list) -> None:
if rows:
stamp = _stamp()
self.conn.execute(insert(self.table), [{**r, **stamp} for r in rows])
def put(self, row: dict) -> None:
"""Create or change by key; the whole row is in VALUES, so it is the row that results."""
upsert(self.conn, self.table, [row], self.key)
def set(self, key: tuple, **values) -> None:
"""UPDATE by primary key."""
cond = and_(*(self.table.c[k] == v for k, v in zip(self.key, key)))
self.conn.execute(update(self.table).where(cond).values(**values, updated_at=_stamp()["updated_at"]))
# ── masters ────────────────────────────────────────────────────────────────────────────────────────────────────────
class Companies(Store):
table, key = t.gl_company, ("company_code",)
def default(self) -> Optional[dict]:
rows = self.rows(t.gl_company.c.is_default.is_(True))
return rows[0] if rows else None
def all(self) -> list:
return self.rows()
class Charts(Store):
table, key = t.gl_chart, ("chart_id",)
class Accounts(Store):
table, key = t.gl_account, ("account_id",)
def by_code(self, chart_id: str, code: str) -> Optional[dict]:
r = self.rows(t.gl_account.c.chart_id == chart_id, t.gl_account.c.code == code)
return r[0] if r else None
def of_chart(self, chart_id: str) -> list:
return self.rows(t.gl_account.c.chart_id == chart_id, order=[t.gl_account.c.code])
def used(self, account_id: int) -> bool:
"""Has any journal line (or a split line) used the account? A used account's code and type are fixed."""
return self.conn.execute(select(t.gl_journal_line.c.journal_id).where(t.gl_journal_line.c.account_id == account_id)
.limit(1)).first() is not None
def children(self, parent_id: int) -> list:
return [r[0] for r in self.conn.execute(select(t.gl_account_edge.c.child_id)
.where(t.gl_account_edge.c.parent_id == parent_id))]
def edges(self) -> list:
return [(r[0], r[1]) for r in self.conn.execute(select(t.gl_account_edge.c.parent_id, t.gl_account_edge.c.child_id))]
def set_edge(self, parent_id: int, child_id: int) -> None:
self.conn.execute(t.gl_account_edge.delete().where(t.gl_account_edge.c.child_id == child_id))
if parent_id is not None:
upsert(self.conn, t.gl_account_edge, [{"parent_id": parent_id, "child_id": child_id}], ("parent_id", "child_id"))
class AccountMap(Store):
table, key = t.gl_account_map, ("chart_id", "purpose", "qualifier", "valid_from")
def rows_of(self, chart_id: str) -> list:
return self.rows(t.gl_account_map.c.chart_id == chart_id,
order=[t.gl_account_map.c.purpose, t.gl_account_map.c.qualifier, t.gl_account_map.c.valid_from])
def account(self, chart_id: str, purpose: str, qualifier: str, on: date) -> Optional[int]:
"""The account in force on a day: the latest row on or before it."""
q = (select(t.gl_account_map.c.account_id)
.where(t.gl_account_map.c.chart_id == chart_id, t.gl_account_map.c.purpose == purpose,
t.gl_account_map.c.qualifier == qualifier, t.gl_account_map.c.valid_from <= on)
.order_by(t.gl_account_map.c.valid_from.desc()).limit(1))
r = self.conn.execute(q).first()
return int(r[0]) if r and r[0] is not None else None
def has_any(self, chart_id: str, purpose: str, qualifier: str) -> bool:
return self.conn.execute(select(t.gl_account_map.c.purpose).where(
t.gl_account_map.c.chart_id == chart_id, t.gl_account_map.c.purpose == purpose,
t.gl_account_map.c.qualifier == qualifier).limit(1)).first() is not None
class GroupMap(Store):
table, key = t.gl_group_account_map, ("account_id", "valid_from")
def of(self, account_id: int, on: date) -> Optional[int]:
q = (select(t.gl_group_account_map.c.group_account_id)
.where(t.gl_group_account_map.c.account_id == account_id, t.gl_group_account_map.c.valid_from <= on)
.order_by(t.gl_group_account_map.c.valid_from.desc()).limit(1))
r = self.conn.execute(q).first()
return int(r[0]) if r and r[0] is not None else None
def history(self, account_id: int) -> list:
return self.rows(t.gl_group_account_map.c.account_id == account_id)
def accounts_of(self, group_account_id: int, on: date) -> list:
"""Every account that reports under the group account on a day (its mapping IN FORCE then, not any older one)."""
ids = {int(r[0]) for r in self.conn.execute(select(t.gl_group_account_map.c.account_id).where(
t.gl_group_account_map.c.group_account_id == group_account_id, t.gl_group_account_map.c.valid_from <= on))}
return sorted(a for a in ids if self.of(a, on) == int(group_account_id))
class AltAccounts(Store):
table, key = t.gl_alt_account, ("company_code", "account_id", "valid_from")
def of(self, company_code: str, account_id: int, on: date) -> Optional[dict]:
rows = self.rows(t.gl_alt_account.c.company_code == company_code, t.gl_alt_account.c.account_id == account_id,
t.gl_alt_account.c.valid_from <= on)
return rows[-1] if rows else None
def history(self, company_code: str, account_id: int) -> list:
return self.rows(t.gl_alt_account.c.company_code == company_code, t.gl_alt_account.c.account_id == account_id)
class Ledgers(Store):
table, key = t.gl_ledger, ("ledger_id",)
def leading(self) -> Optional[dict]:
rows = self.rows(t.gl_ledger.c.kind == "leading")
return rows[0] if rows else None
class CompanyLedgers(Store):
table, key = t.gl_company_ledger, ("company_code", "ledger_id")
def of_company(self, company_code: str) -> list:
return self.rows(t.gl_company_ledger.c.company_code == company_code)
def assigned(self, company_code: str, fiscal_year: int) -> list:
"""The ledger ids the company keeps in a financial year (by number)."""
out = []
for r in self.of_company(company_code):
lo = dates.fy_number(r["from_fiscal_year"]) if r["from_fiscal_year"] else 0
hi = dates.fy_number(r["until_fiscal_year"]) if r["until_fiscal_year"] else 10 ** 6
if lo <= fiscal_year <= hi:
out.append(r["ledger_id"])
return sorted(out)
class Dimensions:
"""Segments, profit centres, cost centres and their dated assignments."""
def __init__(self, conn):
self.conn = conn
self.segments = _Keyed(conn, t.gl_segment, ("segment",))
self.profit_centres = _Keyed(conn, t.gl_profit_centre, ("profit_centre",))
self.pc_segments = _Keyed(conn, t.gl_profit_centre_segment, ("profit_centre", "valid_from"))
self.cost_centres = _Keyed(conn, t.gl_cost_centre, ("cost_centre",))
self.cc_assignments = _Keyed(conn, t.gl_cost_centre_assignment, ("cost_centre", "valid_from"))
def segment_of(self, profit_centre: str, on: date) -> str:
rows = [r for r in self.pc_segments.rows(t.gl_profit_centre_segment.c.profit_centre == profit_centre)
if r["valid_from"] <= on]
return rows[-1]["segment"] if rows else ""
def profit_centre_of_cost_centre(self, cost_centre: str, on: date) -> str:
rows = [r for r in self.cc_assignments.rows(t.gl_cost_centre_assignment.c.cost_centre == cost_centre)
if r["valid_from"] <= on]
return rows[-1]["profit_centre"] if rows else ""
class _Keyed(Store):
def __init__(self, conn, table, key):
super().__init__(conn)
self.table, self.key = table, key
# ── settings and access ────────────────────────────────────────────────────────────────────────────────────────────
class SettingsRepo(Store):
"""Dated setting rows: the value in force on a day is the row with the latest valid_from on or before it."""
table, key = t.gl_settings, ("setting_key", "scope", "valid_from")
def history(self, key: Optional[str] = None, scope: Optional[str] = None) -> list:
"""The setting rows (of a key, of a scope), ordered by key, scope, valid_from. Every row is read in ONE query once per
request and filtered here (a write through this repository forgets them)."""
if self.__dict__.get("_all") is None:
self.__dict__["_all"] = [{**r, "value": json.loads(r["value"]) if r["value"] is not None else None}
for r in self.rows(order=[t.gl_settings.c.setting_key, t.gl_settings.c.scope, t.gl_settings.c.valid_from])]
return [dict(r) for r in self.__dict__["_all"]
if (not key or r["setting_key"] == key) and (scope is None or r["scope"] == scope)]
def add(self, row: dict) -> None:
self.__dict__["_all"] = None
super().add(row)
def add_many(self, rows: list) -> None:
self.__dict__["_all"] = None
super().add_many(rows)
def put(self, row: dict) -> None:
self.__dict__["_all"] = None
super().put(row)
def set(self, key: tuple, **values) -> None:
self.__dict__["_all"] = None
super().set(key, **values)
def in_force(self, on: date, company_code: str = "") -> Settings:
"""The settings on a day: the bucket's row in force, overridden by the company's own row for a per-company setting."""
bucket: dict = {}
company: dict = {}
for r in self.history(): # ordered by key, scope, valid_from: the latest row on or before the day wins
if r["valid_from"] > on or r["setting_key"] not in SETTINGS:
continue
if r["scope"] == "":
bucket[r["setting_key"]] = r["value"]
elif r["scope"] == company_code:
company[r["setting_key"]] = r["value"]
return Settings.of({**bucket, **company})
def has_any(self, key: str, scope: str = "") -> bool:
return bool(self.history(key, scope))
def add_row(self, key: str, scope: str, valid_from: date, value, by: str, at: datetime) -> None:
"""A plain INSERT: a row is never edited; the same key, scope and day twice fails on the primary key."""
self.add({"setting_key": key, "scope": scope, "valid_from": valid_from, "value": json.dumps(value), "set_by": by,
"set_at": utc(at)})
class CompanyAccess(Store):
table, key = t.gl_company_access, ("principal_sub", "company_code", "valid_from")
def allowed(self, sub: str, email: str, on: date) -> Optional[set]:
"""None when the person has no restriction (no row has taken effect for them), else the company codes in force today."""
mine = [r for r in self.rows() if r["principal_sub"] == sub or (email and (r["principal_email"] or "").lower() == email.lower())]
taken = [r for r in mine if r["valid_from"] <= on]
if not taken:
return None
return {r["company_code"] for r in taken if r["valid_until"] is None or r["valid_until"] >= on}
# ── fx ─────────────────────────────────────────────────────────────────────────────────────────────────────────────
class FxRates(Store):
table, key = t.gl_fx_rate, ("from_ccy", "to_ccy", "rate_type", "rate_date")
def in_force(self, from_ccy: str, to_ccy: str, rate_type: str, on: date) -> Optional[dict]:
q = (select(t.gl_fx_rate).where(t.gl_fx_rate.c.from_ccy == from_ccy, t.gl_fx_rate.c.to_ccy == to_ccy,
t.gl_fx_rate.c.rate_type == rate_type, t.gl_fx_rate.c.rate_date <= on)
.order_by(t.gl_fx_rate.c.rate_date.desc()).limit(1))
return _d(self.conn.execute(q).first())
def on_date(self, from_ccy: str, to_ccy: str, rate_type: str, day: date) -> Optional[dict]:
return self.get(from_ccy, to_ccy, rate_type, day)
def listing(self, from_ccy: Optional[str] = None, to_ccy: Optional[str] = None) -> list:
conds = []
if from_ccy:
conds.append(t.gl_fx_rate.c.from_ccy == from_ccy)
if to_ccy:
conds.append(t.gl_fx_rate.c.to_ccy == to_ccy)
return self.rows(*conds, order=[t.gl_fx_rate.c.from_ccy, t.gl_fx_rate.c.rate_type, t.gl_fx_rate.c.rate_date])
# ── fiscal years, periods ──────────────────────────────────────────────────────────────────────────────────────────
class FiscalYears(Store):
table, key = t.gl_fiscal_year, ("company_code", "fiscal_year")
def of_company(self, company_code: str) -> list:
return self.rows(t.gl_fiscal_year.c.company_code == company_code, order=[t.gl_fiscal_year.c.starts_on])
class Periods(Store):
table, key = t.gl_period, ("company_code", "period_id")
def for_date(self, company_code: str, on: date) -> Optional[dict]:
r = self.rows(t.gl_period.c.company_code == company_code, t.gl_period.c.starts_on <= on, t.gl_period.c.ends_on >= on)
return r[0] if r else None
def of_company(self, company_code: str, fiscal_year: Optional[str] = None) -> list:
conds = [t.gl_period.c.company_code == company_code]
if fiscal_year:
conds.append(t.gl_period.c.fiscal_year == fiscal_year)
return self.rows(*conds, order=[t.gl_period.c.period_id])
def events(self, company_code: str, period_id: str) -> list:
return _Keyed(self.conn, t.gl_period_event, ("event_id",)).rows(
t.gl_period_event.c.company_code == company_code, t.gl_period_event.c.period_id == period_id,
order=[t.gl_period_event.c.event_at])
def log(self, company_code: str, period_id: str, event: str, reason: str, by: str, sub: str, at: datetime) -> None:
_Keyed(self.conn, t.gl_period_event, ("event_id",)).add({
"event_id": uuid.uuid4().hex, "company_code": company_code, "period_id": period_id, "event": event,
"reason": reason, "by_email": by, "by_sub": sub, "event_at": utc(at)})
def touch(self, company_code: str, period_id: str, at: datetime) -> None:
"""Every posting writes the period row in its own transaction (last_posted_at), so a close racing a posting
conflicts on this row and the loser retries: the lock is a guard on the one write path, not a constraint."""
self.set((company_code, period_id), last_posted_at=utc(at))
# ── numbers and the chain ──────────────────────────────────────────────────────────────────────────────────────────
class Numbers(Store):
table, key = t.gl_number_series, ("company_code", "book", "fiscal_year", "entry_no")
def claim(self, company_code: str, book: str, fiscal_year: str, journal_id: int, at: datetime, floor: int = 0) -> int:
"""The next entry number of a company's book and year: one above the highest ever claimed (never reusing or filling a
gap below it). A PLAIN INSERT: the key is the uniqueness guard, so two postings racing for one number cannot both
commit - the loser fails 23505 and the use case retries with the next."""
top = self.conn.execute(select(func.max(t.gl_number_series.c.entry_no)).where(
t.gl_number_series.c.company_code == company_code, t.gl_number_series.c.book == book,
t.gl_number_series.c.fiscal_year == fiscal_year)).scalar() or 0
n = max(int(top), int(floor)) + 1
self.add({"company_code": company_code, "book": book, "fiscal_year": fiscal_year, "entry_no": n, "journal_id": journal_id,
"claimed_at": utc(at)})
return n
def high_water(self, company_code: str, book: str, fiscal_year: str) -> int:
return int(self.conn.execute(select(func.max(t.gl_number_series.c.entry_no)).where(
t.gl_number_series.c.company_code == company_code, t.gl_number_series.c.book == book,
t.gl_number_series.c.fiscal_year == fiscal_year)).scalar() or 0)
class Chain(Store):
table, key = t.gl_chain_claim, ("company_code", "chain_seq")
def claim(self, company_code: str, journal_id: int, at: datetime) -> tuple:
"""(chain_seq, prev_hash): the next link of the company's chain, claimed with a plain INSERT like a number."""
top = self.conn.execute(select(func.max(t.gl_chain_claim.c.chain_seq)).where(
t.gl_chain_claim.c.company_code == company_code)).scalar() or 0
seq = int(top) + 1
prev = GENESIS
if top:
r = self.conn.execute(select(t.gl_journal.c.row_hash).where(
t.gl_journal.c.company_code == company_code, t.gl_journal.c.chain_seq == int(top))).first()
prev = r[0] if r and r[0] else GENESIS
self.add({"company_code": company_code, "chain_seq": seq, "journal_id": journal_id, "claimed_at": utc(at)})
return seq, prev
# ── journals ───────────────────────────────────────────────────────────────────────────────────────────────────────
class Journals(Store):
table, key = t.gl_journal, ("journal_id",)
def lines(self, journal_id: int) -> list:
return [dict(r._mapping) for r in self.conn.execute(
select(t.gl_journal_line).where(t.gl_journal_line.c.journal_id == journal_id).order_by(t.gl_journal_line.c.line_no))]
def splits(self, journal_id: int) -> list:
return [dict(r._mapping) for r in self.conn.execute(
select(t.gl_split_line).where(t.gl_split_line.c.journal_id == journal_id)
.order_by(t.gl_split_line.c.line_no, t.gl_split_line.c.split_no))]
def add_lines(self, rows: list) -> None:
if rows:
stamp = _stamp()
self.conn.execute(insert(t.gl_journal_line), [{**r, **stamp} for r in rows])
def add_splits(self, rows: list) -> None:
if rows:
stamp = _stamp()
self.conn.execute(insert(t.gl_split_line), [{**r, **stamp} for r in rows])
def put_lines(self, rows: list) -> None:
"""Lines of a PROPOSED journal are written again when it posts (the purposes resolved on the posting day)."""
upsert(self.conn, t.gl_journal_line, [{**r} for r in rows], ("journal_id", "line_no"))
def audit(self, journal_id: int, event: str, by: str, sub: str, at: datetime) -> None:
_Keyed(self.conn, t.gl_journal_audit, ("audit_id",)).add({
"audit_id": uuid.uuid4().hex, "journal_id": journal_id, "event": event, "by_email": by, "by_sub": sub,
"event_at": utc(at)})
def trail(self, journal_id: int) -> list:
return _Keyed(self.conn, t.gl_journal_audit, ("audit_id",)).rows(
t.gl_journal_audit.c.journal_id == journal_id, order=[t.gl_journal_audit.c.event_at])
def reversal_of(self, journal_id: int) -> Optional[dict]:
r = self.rows(t.gl_journal.c.reverses_journal_id == journal_id)
return r[0] if r else None
def by_ic_document(self, ic_document_id: int) -> list:
return self.rows(t.gl_journal.c.ic_document_id == ic_document_id)
def pending_reversals(self, company_code: str, upto: Optional[date] = None) -> list:
return self.rows(t.gl_journal.c.company_code == company_code, t.gl_journal.c.reversal_status == "pending",
order=[t.gl_journal.c.journal_date, t.gl_journal.c.journal_id])
def listing(self, company_codes: Optional[list] = None, status: Optional[str] = None, period_id: Optional[str] = None,
source_module: Optional[str] = None, book: Optional[str] = None, after: int = 0, limit: int = 50) -> list:
j = t.gl_journal.c
conds = [j.journal_id > after]
if company_codes is not None:
conds.append(j.company_code.in_(company_codes))
for col, v in ((j.status, status), (j.period_id, period_id), (j.source_module, source_module), (j.book, book)):
if v:
conds.append(col == v)
return self.rows(*conds, order=[j.journal_id], limit=limit)
def proposed_in(self, company_code: str, period_id: str) -> list:
j = t.gl_journal.c
return self.rows(j.company_code == company_code, j.period_id == period_id, j.status == "proposed")
def posted_since(self, company_code: str, period_id: str, since: Optional[datetime]) -> int:
j = t.gl_journal.c
q = select(func.count()).select_from(t.gl_journal).where(j.company_code == company_code, j.period_id == period_id,
j.status.in_(("posted", "reversed")))
if since is not None:
q = q.where(j.posted_at > utc(since))
return int(self.conn.execute(q).scalar() or 0)
class Ledger:
"""Everything derived from split lines: balances, the trial balance, an account's lines. Never incremented."""
def __init__(self, conn):
self.conn = conn
def _ledger_cond(self, ledger_id: Optional[str]):
s = t.gl_split_line.c
return or_(s.ledger_id.is_(None), s.ledger_id == ledger_id) if ledger_id else s.ledger_id.is_(None)
def sums(self, company_code: str, ledger_id: str, *, period_id: Optional[str] = None, upto: Optional[date] = None,
since: Optional[date] = None, account_ids: Optional[list] = None, group_by: tuple = ("account_id",),
adjustments: bool = True, extra: Optional[list] = None) -> list:
"""SUM of split lines per group: [{...group..., debit, credit, f_debit, f_credit}] in the functional currency
(debit/credit are the transaction currency). Common journals (ledger_id NULL) count in every ledger."""
s = t.gl_split_line.c
cols = [s[g] for g in group_by]
q = select(*cols,
type_coerce(func.sum(s.debit), MONEY).label("debit"), type_coerce(func.sum(s.credit), MONEY).label("credit"),
type_coerce(func.sum(s.functional_debit), MONEY).label("f_debit"),
type_coerce(func.sum(s.functional_credit), MONEY).label("f_credit"))
conds = [s.company_code == company_code, self._ledger_cond(ledger_id)]
if period_id:
conds.append(s.period_id == period_id)
if upto:
conds.append(s.journal_date <= upto)
if since:
conds.append(s.journal_date >= since)
if account_ids is not None:
conds.append(s.account_id.in_(list(account_ids)))
if not adjustments:
conds.append(s.adjustment.is_(False))
conds += extra or []
q = q.where(and_(*conds)).group_by(*cols)
out = []
for r in self.conn.execute(q):
m = r._mapping
out.append({**{g: m[g] for g in group_by}, "debit": _money(m["debit"]), "credit": _money(m["credit"]),
"f_debit": _money(m["f_debit"]), "f_credit": _money(m["f_credit"])})
return out
def lines(self, company_code: str, ledger_id: str, account_id, since: Optional[date], upto: Optional[date]) -> list:
"""An account's split lines (account_id: one id, or a list: every account mapped to a group account)."""
s = t.gl_split_line.c
j = t.gl_journal.c
conds = [s.company_code == company_code, self._ledger_cond(ledger_id),
s.account_id.in_(list(account_id)) if isinstance(account_id, (list, tuple)) else s.account_id == account_id]
if since:
conds.append(s.journal_date >= since)
if upto:
conds.append(s.journal_date <= upto)
q = (select(s.journal_id, s.account_id, s.line_no, s.split_no, s.journal_date, s.debit, s.credit, s.functional_debit, s.functional_credit,
s.currency, s.profit_centre, s.segment, s.partner_company, s.bp_id, s.is_clearing, j.entry_no, j.book, j.fiscal_year,
j.source_module, j.source_document, j.event, j.memo, j.period_id)
.select_from(t.gl_split_line.join(t.gl_journal, t.gl_journal.c.journal_id == s.journal_id))
.where(and_(*conds)).order_by(s.journal_date, s.journal_id, s.line_no, s.split_no))
return [dict(r._mapping) for r in self.conn.execute(q)]
def put_balances(self, rows: list) -> None:
upsert(self.conn, t.gl_balance, rows, ("company_code", "ledger_id", "account_id", "period_id", "profit_centre", "segment",
"partner_company"), insert_only=("created_at",))
def balances(self, company_code: str, ledger_id: str, period_id: Optional[str] = None, account_id: Optional[int] = None) -> list:
b = t.gl_balance.c
conds = [b.company_code == company_code, b.ledger_id == ledger_id]
if period_id:
conds.append(b.period_id == period_id)
if account_id is not None:
conds.append(b.account_id == account_id)
return [dict(r._mapping) for r in self.conn.execute(select(t.gl_balance).where(and_(*conds)).order_by(
b.period_id, b.account_id, b.profit_centre, b.segment, b.partner_company))]
def passive_weights(self, against_document: str, account_id: int) -> dict:
"""How the document a payment clears was split on that account: {profit centre: |net|} from its split lines."""
s = t.gl_split_line.c
j = t.gl_journal.c
q = (select(s.profit_centre, type_coerce(func.sum(s.functional_debit - s.functional_credit), MONEY))
.select_from(t.gl_split_line.join(t.gl_journal, j.journal_id == s.journal_id))
.where(j.source_document == against_document, j.source_module != "gl", s.account_id == account_id, s.is_clearing.is_(False))
.group_by(s.profit_centre))
return {r[0]: abs(_money(r[1])) for r in self.conn.execute(q) if r[1] is not None and _money(r[1]) != 0}
def carry(self, company_code: str, ledger_id: str, fiscal_year: str) -> list:
c = t.gl_carry_forward.c
return [dict(r._mapping) for r in self.conn.execute(select(t.gl_carry_forward).where(
c.company_code == company_code, c.ledger_id == ledger_id, c.fiscal_year == fiscal_year))]
def put_carry(self, rows: list) -> None:
upsert(self.conn, t.gl_carry_forward, rows, ("company_code", "ledger_id", "fiscal_year", "account_id", "profit_centre",
"segment", "partner_company"), insert_only=("created_at",))
class Repository:
"""Every repository on one connection: what `use_cases.repository()` yields."""
def __init__(self, conn):
self.conn = conn
self.companies = Companies(conn)
self.charts = Charts(conn)
self.accounts = Accounts(conn)
self.account_map = AccountMap(conn)
self.group_map = GroupMap(conn)
self.alt_accounts = AltAccounts(conn)
self.ledgers = Ledgers(conn)
self.company_ledgers = CompanyLedgers(conn)
self.dims = Dimensions(conn)
self.settings = SettingsRepo(conn)
self.access = CompanyAccess(conn)
self.fx = FxRates(conn)
self.years = FiscalYears(conn)
self.periods = Periods(conn)
self.numbers = Numbers(conn)
self.chain = Chain(conn)
self.journals = Journals(conn)
self.ledger = Ledger(conn)
self.master_audit = _Keyed(conn, t.gl_master_audit, ("audit_id",))
def audit_master(self, entity: str, natural_key: str, before: Optional[dict], after: Optional[dict], by: str, sub: str,
at: datetime) -> None:
"""GoBD: every change to master data and settings, with before and after (append only)."""
def enc(v):
return json.dumps(v, default=str, sort_keys=True) if v is not None else None
self.master_audit.add({"audit_id": uuid.uuid4().hex, "entity": entity, "natural_key": natural_key,
"before_json": enc(before), "after_json": enc(after), "by_email": by, "by_sub": sub,
"event_at": utc(at)})
def control_documents(repo, company_code: str, subtype: str, foreign_only: bool = False, functional: str = "") -> list:
"""Distinct source documents of posted journals with a line on a control (receivable / payable) account: the invoices ar
posted, the bills purchasing posted. [(source_document, journal row, account_id)], the first journal of each document."""
j, ln, a = t.gl_journal.c, t.gl_journal_line.c, t.gl_account.c
q = (select(t.gl_journal, ln.account_id.label("control_account_id"))
.select_from(t.gl_journal.join(t.gl_journal_line, ln.journal_id == j.journal_id).join(t.gl_account, a.account_id == ln.account_id))
.where(j.company_code == company_code, j.status.in_(("posted", "reversed")), a.subtype == subtype,
j.source_module != "gl", j.reverses_journal_id.is_(None))
.order_by(j.journal_date, j.journal_id))
if foreign_only:
q = q.where(j.currency != functional)
seen, out = set(), []
for r in repo.conn.execute(q):
m = dict(r._mapping)
if m["source_document"] in seen:
continue
seen.add(m["source_document"])
out.append((m["source_document"], m, m["control_account_id"]))
return out