forked from rohitdash08/FinMind
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathindex_report.py
More file actions
91 lines (75 loc) · 2.77 KB
/
Copy pathindex_report.py
File metadata and controls
91 lines (75 loc) · 2.77 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
"""
Database index report utility (Issue #128).
Run with: flask index-report
Reports: current indexes, missing recommended indexes, table sizes.
Helps track index health in production.
"""
from __future__ import annotations
import click
from flask import Flask
from flask.cli import with_appcontext
from ..extensions import db
# Expected indexes — checked at startup in dev/test to catch regressions
REQUIRED_INDEXES = [
"idx_expenses_user_spent_at",
"idx_expenses_user_category",
"idx_expenses_user_type_date",
"idx_expenses_user_month",
"idx_bills_user_due",
"idx_bills_user_active_due",
"idx_reminders_due",
"idx_reminders_pending_dispatch",
"idx_recurring_expenses_user_start",
"idx_recurring_expenses_active",
"idx_categories_user_name",
"idx_audit_logs_user_created",
"idx_audit_logs_action_created",
]
def get_existing_indexes(engine) -> list[str]:
"""Query pg_indexes for all FinMind indexes."""
if engine.dialect.name != "postgresql":
return []
with engine.connect() as conn:
result = conn.execute(
db.text(
"""
SELECT indexname FROM pg_indexes
WHERE schemaname = 'public'
ORDER BY indexname
"""
)
)
return [row[0] for row in result]
def check_missing_indexes(engine) -> list[str]:
"""Return list of recommended indexes not yet present."""
existing = set(get_existing_indexes(engine))
return [idx for idx in REQUIRED_INDEXES if idx not in existing]
def register_index_cli(app: Flask) -> None:
"""Register the flask index-report CLI command."""
@app.cli.command("index-report")
@with_appcontext
def index_report():
"""Print a report of database indexes and missing recommendations."""
engine = db.engine
if engine.dialect.name != "postgresql":
click.echo("Index report only supported on PostgreSQL.")
return
existing = get_existing_indexes(engine)
missing = check_missing_indexes(engine)
click.echo(f"\n{'='*60}")
click.echo("FinMind Database Index Report")
click.echo(f"{'='*60}")
click.echo(f"\nTotal indexes: {len(existing)}")
for idx in sorted(existing):
status = "✅" if idx in REQUIRED_INDEXES else " "
click.echo(f" {status} {idx}")
if missing:
click.echo(f"\n⚠️ Missing recommended indexes ({len(missing)}):")
for idx in missing:
click.echo(f" ❌ {idx}")
click.echo(
"\nRun migration 007_db_indexing.sql to add missing indexes."
)
else:
click.echo("\n✅ All recommended indexes are present.")
click.echo(f"{'='*60}\n")