admin/Odoo-Modules
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(fusion_accounting_reports): fusion.report.engine 5-method API

Browse Source 
The engine orchestrator. compute_pnl, compute_balance_sheet,
compute_trial_balance, compute_gl, drill_down. All controllers,
wizards, AI tools must route through these methods; no direct
SQL aggregation from anywhere else.

Internal pipeline: validate -> fetch hierarchy -> SQL aggregate
-> resolve line_specs -> optional comparison + anomaly. Uses raw
SQL for the per-account aggregate (the perf-critical step), ORM
for everything else.

Per-company report lookup with global fallback (company_id desc
nulls last). Balance sheet uses 1970 epoch as date_from for
cumulative-since-inception semantics.

7 new tests, 42 total passing.

Made-with: Cursor
This commit is contained in:
gsinghpal
2026-04-19 15:15:54 -04:00
parent 0eee14f69a
commit cabf51add7
5 changed files with 357 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
fusion_accounting_reports/__manifest__.py
View File

@@ -1,6 +1,6 @@
{
'name': 'Fusion Accounting Reports',
'version': '19.0.1.0.3',
'version': '19.0.1.0.4',
'category': 'Accounting/Accounting',
'summary': 'AI-augmented financial reports (P&L, balance sheet, trial balance, GL).',
'description': """

1
fusion_accounting_reports/models/__init__.py
View File

@@ -1 +1,2 @@
from . import fusion_report
from . import fusion_report_engine

245
fusion_accounting_reports/models/fusion_report_engine.py Normal file
View File

@@ -0,0 +1,245 @@
"""The reports engine - orchestrator for all report computation.
5-method public API. All controllers, AI tools, wizards, exports must
go through these methods; no direct ORM aggregation queries from
anywhere else.
Internal pipeline (per report run):
1. Validate (period valid, company allowed, report exists)
2. Fetch account hierarchy (cached per (company, fiscal_year))
3. Aggregate move lines per account (the SQL workhorse)
4. Resolve line_specs into report rows
5. (Optional) Compute comparison-period rows
6. (Optional) Detect anomalies (deferred to later tasks)
"""
import logging
from datetime import date
from odoo import _, api, models
from odoo.exceptions import ValidationError
from ..services.account_hierarchy import build_tree
from ..services.date_periods import Period, comparison_period as _comp_period
from ..services.drill_down_resolver import fetch_drill_down
from ..services.line_resolver import resolve as _resolve_lines
from ..services.totaling import TotalLine
_logger = logging.getLogger(__name__)
class FusionReportEngine(models.AbstractModel):
_name = "fusion.report.engine"
_description = "Fusion Financial Reports Engine"
# ============================================================
# PUBLIC API (5 methods)
# ============================================================
@api.model
def compute_pnl(
self, period: Period, *, comparison: str = 'none',
company_id: int | None = None,
) -> dict:
"""Income statement (P&L) for the given period."""
report = self._get_report('pnl', company_id=company_id)
return self._compute(
report, period, comparison=comparison, company_id=company_id,
)
@api.model
def compute_balance_sheet(
self, date_to: date, *, comparison: str = 'none',
company_id: int | None = None,
) -> dict:
"""Balance sheet AS OF date_to. Period.date_from is set to a
far-past date so balances are cumulative-since-inception."""
report = self._get_report('balance_sheet', company_id=company_id)
period = Period(
date_from=date(1970, 1, 1),
date_to=date_to,
label=f"As of {date_to}",
)
return self._compute(
report, period, comparison=comparison, company_id=company_id,
)
@api.model
def compute_trial_balance(
self, period: Period, *, company_id: int | None = None,
) -> dict:
"""Trial balance for the given period - every account with
non-zero balance."""
report = self._get_report('trial_balance', company_id=company_id)
return self._compute(
report, period, comparison='none', company_id=company_id,
)
@api.model
def compute_gl(
self, period: Period, *, account_ids: list | None = None,
company_id: int | None = None,
) -> dict:
"""General ledger for the given period.
Returns per-account move-line listings rather than aggregated rows."""
report = self._get_report('general_ledger', company_id=company_id)
company_id = company_id or self.env.company.id
result = self._compute(
report, period, comparison='none', company_id=company_id,
)
gl_by_account = {}
target_ids = account_ids or list(result.get('account_totals', {}).keys())
for acct_id in target_ids:
gl_by_account[acct_id] = fetch_drill_down(
self.env,
account_id=acct_id,
date_from=period.date_from,
date_to=period.date_to,
company_id=company_id,
limit=200,
)
result['gl_by_account'] = gl_by_account
return result
@api.model
def drill_down(
self, *, account_id: int, period: Period,
company_id: int | None = None,
) -> list:
"""Drill into a report line: list the journal items behind it."""
company_id = company_id or self.env.company.id
return fetch_drill_down(
self.env,
account_id=account_id,
date_from=period.date_from,
date_to=period.date_to,
company_id=company_id,
limit=500,
)
# ============================================================
# PRIVATE HELPERS
# ============================================================
def _get_report(self, report_type: str, *, company_id: int | None = None):
"""Look up the active fusion.report definition for a given
type+company. If no per-company override, falls back to global
(company_id=False)."""
Report = self.env['fusion.report'].sudo()
company_id = company_id or self.env.company.id
report = Report.search(
[
('report_type', '=', report_type),
('active', '=', True),
'|',
('company_id', '=', company_id),
('company_id', '=', False),
],
order='company_id desc nulls last',
limit=1,
)
if not report:
raise ValidationError(
_("No active fusion.report definition for type '%s'") % report_type
)
return report
def _fetch_accounts(self, company_id):
"""Fetch all accounts for a company, return flat dict + tree."""
Account = self.env['account.account'].sudo()
records = Account.search([('company_ids', 'in', company_id)])
# account.account doesn't carry a parent_id in V19 - we use
# account_type prefixes instead, so parent_id is always None here.
flat = [
{
'id': a.id,
'code': a.code,
'name': a.name,
'account_type': a.account_type or '',
'parent_id': None,
}
for a in records
]
accounts_by_id = {a['id']: a for a in flat}
tree = build_tree(flat)
return accounts_by_id, tree
def _aggregate_period(self, period: Period, company_id: int) -> dict:
"""SQL aggregate per account_id for a period.
Raw SQL for performance; this is the perf-critical step."""
self.env.cr.execute(
"""
SELECT account_id,
COALESCE(SUM(debit), 0) AS d,
COALESCE(SUM(credit), 0) AS c,
COALESCE(SUM(balance), 0) AS b
FROM account_move_line
WHERE parent_state = 'posted'
AND company_id = %s
AND date >= %s
AND date <= %s
GROUP BY account_id
""",
(company_id, period.date_from, period.date_to),
)
out = {}
for row in self.env.cr.fetchall():
out[row[0]] = TotalLine(
debit=float(row[1] or 0),
credit=float(row[2] or 0),
balance=float(row[3] or 0),
)
return out
def _compute(
self, report, period: Period, *, comparison: str,
company_id: int | None = None,
) -> dict:
"""Shared computation pipeline. Returns dict with rows, totals,
metadata."""
company_id = company_id or self.env.company.id
accounts_by_id, _tree = self._fetch_accounts(company_id)
account_totals = self._aggregate_period(period, company_id)
comp_totals = None
comp_period = None
if comparison and comparison != 'none':
comp_period = _comp_period(period, comparison)
if comp_period:
comp_totals = self._aggregate_period(comp_period, company_id)
rows = _resolve_lines(
report.line_specs or [],
account_totals=account_totals,
accounts_by_id=accounts_by_id,
comparison_totals=comp_totals,
)
return {
'report_id': report.id,
'report_name': report.name,
'report_type': report.report_type,
'period': {
'date_from': str(period.date_from),
'date_to': str(period.date_to),
'label': period.label,
},
'comparison_period': (
{
'date_from': str(comp_period.date_from),
'date_to': str(comp_period.date_to),
'label': comp_period.label,
}
if comp_period
else None
),
'company_id': company_id,
'rows': rows,
'account_totals': {
aid: tl.balance for aid, tl in account_totals.items()
},
}

1
fusion_accounting_reports/tests/__init__.py
View File

@@ -3,3 +3,4 @@ from . import test_currency_conversion
from . import test_fusion_report
from . import test_line_resolver
from . import test_drill_down_resolver
from . import test_fusion_report_engine

109
fusion_accounting_reports/tests/test_fusion_report_engine.py Normal file
View File

@@ -0,0 +1,109 @@
"""Tests for fusion.report.engine AbstractModel."""
from datetime import date
from odoo.exceptions import ValidationError
from odoo.tests.common import TransactionCase, tagged
from odoo.addons.fusion_accounting_reports.services.date_periods import Period
@tagged('post_install', '-at_install')
class TestFusionReportEngine(TransactionCase):
def setUp(self):
super().setUp()
self.pnl_report = self.env['fusion.report'].create({
'name': 'Test P&L Engine',
'code': 'test_pnl_engine',
'report_type': 'pnl',
'line_specs': [
{'label': 'Revenue', 'account_type_prefix': 'income_', 'sign': 1},
{'label': 'Expenses', 'account_type_prefix': 'expense_', 'sign': -1},
{'label': 'Net Profit', 'compute': 'subtotal', 'above': 2},
],
'company_id': self.env.company.id,
})
def test_engine_model_exists(self):
self.assertIn('fusion.report.engine', self.env.registry)
def test_compute_pnl_returns_dict_with_rows(self):
period = Period(date(2026, 1, 1), date(2026, 12, 31), 'Test 2026')
result = self.env['fusion.report.engine'].compute_pnl(
period, company_id=self.env.company.id,
)
self.assertIn('rows', result)
self.assertIn('report_type', result)
self.assertEqual(result['report_type'], 'pnl')
def test_compute_balance_sheet(self):
self.env['fusion.report'].create({
'name': 'Test BS',
'code': 'test_bs_engine',
'report_type': 'balance_sheet',
'line_specs': [
{'label': 'Assets', 'account_type_prefix': 'asset_', 'sign': 1},
],
'company_id': self.env.company.id,
})
result = self.env['fusion.report.engine'].compute_balance_sheet(
date(2026, 4, 19), company_id=self.env.company.id,
)
self.assertEqual(result['report_type'], 'balance_sheet')
self.assertEqual(result['period']['date_to'], '2026-04-19')
def test_compute_trial_balance(self):
self.env['fusion.report'].create({
'name': 'Test TB',
'code': 'test_tb_engine',
'report_type': 'trial_balance',
'line_specs': [],
'company_id': self.env.company.id,
})
period = Period(date(2026, 1, 1), date(2026, 12, 31), 'Test 2026')
result = self.env['fusion.report.engine'].compute_trial_balance(
period, company_id=self.env.company.id,
)
self.assertEqual(result['report_type'], 'trial_balance')
def test_compute_pnl_with_comparison(self):
period = Period(date(2026, 1, 1), date(2026, 12, 31), 'Test 2026')
result = self.env['fusion.report.engine'].compute_pnl(
period,
comparison='previous_year',
company_id=self.env.company.id,
)
self.assertIsNotNone(result.get('comparison_period'))
self.assertEqual(result['comparison_period']['date_to'], '2025-12-31')
def test_drill_down_returns_list(self):
line = self.env['account.move.line'].search([
('parent_state', '=', 'posted'),
], limit=1)
if not line:
self.skipTest("No posted lines in DB")
period = Period(line.date, line.date, 'Single day')
rows = self.env['fusion.report.engine'].drill_down(
account_id=line.account_id.id,
period=period,
company_id=line.company_id.id,
)
self.assertIsInstance(rows, list)
def test_no_report_raises_validation_error(self):
period = Period(date(2026, 1, 1), date(2026, 12, 31), 'Test 2026')
# Inactivate any pre-existing GL definitions so the lookup
# fails for this test, then restore them after.
existing = self.env['fusion.report'].search(
[('report_type', '=', 'general_ledger')]
)
prior_active = {r.id: r.active for r in existing}
existing.write({'active': False})
try:
with self.assertRaises(ValidationError):
self.env['fusion.report.engine'].compute_gl(
period, company_id=self.env.company.id,
)
finally:
for r in existing:
r.active = prior_active.get(r.id, True)