Movses AslanyanandClaude Opus 5.5
d52a19965a
Manual INTERNAL spans for the business steps the auto-instrumentation can't name: fraud.score_transaction, sanctions.lookup_watchlist, ledger.settle and insights.aggregate_120d, with ledgerline.* attributes (account, amount, risk score, match outcome). holder_name stays off the spans. Each span starts after its handler's chaos flag reads and injection points, and no chaos-derived values become attributes. The gateway's /api/health/all fan-out now runs under suppress_instrumentation: with the route excluded, its five downstream /healthz calls were each a parentless root trace. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
91 lines
3.4 KiB
Python
91 lines
3.4 KiB
Python
import sqlalchemy as sa
|
|
from opentelemetry import trace
|
|
|
|
# Deliberately the same shape of query the test suite's own EXPLAIN check runs
|
|
# against the schema directly (tests/test_pathologies.py, 6.4) -- kept here
|
|
# only for the honest (non-n_plus_one) path; that test doesn't call this code.
|
|
|
|
_N_PLUS_ONE_ROW_CAP = 1500
|
|
|
|
tracer = trace.get_tracer(__name__)
|
|
|
|
|
|
async def account_exists(session, account_id: str) -> bool:
|
|
result = await session.execute(sa.text("SELECT 1 FROM accounts WHERE id = :id"), {"id": account_id})
|
|
return result.first() is not None
|
|
|
|
|
|
async def compute_summary(session, account_id: str, *, n_plus_one: bool) -> dict:
|
|
with tracer.start_as_current_span(
|
|
"insights.aggregate_120d",
|
|
attributes={"ledgerline.account.id": account_id, "ledgerline.summary.window_days": 120},
|
|
) as span:
|
|
if n_plus_one:
|
|
rows = await _n_plus_one_summary(session, account_id)
|
|
else:
|
|
rows = await _joined_summary(session, account_id)
|
|
|
|
total_count = 0
|
|
total_amount = 0
|
|
by_category: dict = {}
|
|
for r in rows:
|
|
total_count += 1
|
|
total_amount += r["amount_minor"]
|
|
label = r["label"] or "uncategorized"
|
|
entry = by_category.setdefault(label, {"count": 0, "total_amount_minor": 0})
|
|
entry["count"] += 1
|
|
entry["total_amount_minor"] += r["amount_minor"]
|
|
|
|
span.set_attribute("ledgerline.summary.transaction_count", total_count)
|
|
return {
|
|
"account_id": account_id,
|
|
"window_days": 120,
|
|
"transaction_count": total_count,
|
|
"total_amount_minor": total_amount,
|
|
"by_category": by_category,
|
|
}
|
|
|
|
|
|
async def _joined_summary(session, account_id: str) -> list[dict]:
|
|
"""The honest path: a single JOIN, one statement."""
|
|
result = await session.execute(
|
|
sa.text(
|
|
"SELECT t.amount_minor, mc.label FROM transactions t "
|
|
"LEFT JOIN merchant_categories mc ON mc.id = t.category_id "
|
|
"WHERE t.account_id = :aid AND t.created_at >= now() - interval '120 days'"
|
|
),
|
|
{"aid": account_id},
|
|
)
|
|
return [dict(row) for row in result.mappings().all()]
|
|
|
|
|
|
async def _n_plus_one_summary(session, account_id: str) -> list[dict]:
|
|
"""n_plus_one: instead of the JOIN above, fetch rows without one and issue
|
|
a separate category lookup per row -- phase-0-base-project-spec.md §7.
|
|
Capped so the pathology stays a genuine multiplication of statements
|
|
(see tests/test_pathologies.py 6.3) without turning a single request into
|
|
an unbounded number of round trips on a high-volume hot account."""
|
|
result = await session.execute(
|
|
sa.text(
|
|
"SELECT amount_minor, category_id FROM transactions "
|
|
"WHERE account_id = :aid AND created_at >= now() - interval '120 days' "
|
|
f"LIMIT {_N_PLUS_ONE_ROW_CAP}"
|
|
),
|
|
{"aid": account_id},
|
|
)
|
|
tx_rows = result.mappings().all()
|
|
|
|
rows = []
|
|
for tx_row in tx_rows:
|
|
label = None
|
|
if tx_row["category_id"] is not None:
|
|
cat_result = await session.execute(
|
|
sa.text("SELECT label FROM merchant_categories WHERE id = :id"),
|
|
{"id": tx_row["category_id"]},
|
|
)
|
|
cat_row = cat_result.first()
|
|
if cat_row is not None:
|
|
label = cat_row.label
|
|
rows.append({"amount_minor": tx_row["amount_minor"], "label": label})
|
|
return rows
|