"""
Unified Expense Core — guaranteed to match MCP modules.
Provides DB-level operations only. All business logic stays in modules/.
"""
from utils.database import get_conn
from utils.date_utils import parse_date, get_month_range
from typing import List, Dict, Any, Optional
# ============================================================
# INTERNAL HELPERS
# ============================================================
def _row_to_dict(row) -> Dict[str, Any]:
"""Convert a SQLite row/tuple into a dict."""
if hasattr(row, "keys"):
return {k: row[k] for k in row.keys()}
return {
"id": row[0],
"amount": row[1],
"category": row[2],
"subcategory": row[3],
"date": row[4],
"description": row[5],
}
# ============================================================
# CREATE / UPDATE / DELETE
# ============================================================
def add_expense_to_db(amount: float, category: str, subcategory: str,
date: str, description: str) -> Dict[str, Any]:
"""Insert an expense into the DB."""
date_iso = parse_date(date)
conn = get_conn()
cur = conn.cursor()
cur.execute(
"""
INSERT INTO expenses (amount, category, subcategory, date, description)
VALUES (?, ?, ?, ?, ?)
""",
(amount, category, subcategory, date_iso, description),
)
conn.commit()
exp_id = cur.lastrowid
conn.close()
return {
"id": exp_id,
"amount": amount,
"category": category,
"subcategory": subcategory,
"date": date_iso,
"description": description,
}
def update_expense_in_db(expense_id: int, **updates) -> Dict[str, Any]:
"""Update any subset of fields for an expense."""
allowed = {"amount", "category", "subcategory", "date", "description"}
fields = []
params = []
for key, value in updates.items():
if key not in allowed:
continue
if key == "date":
value = parse_date(value)
fields.append(f"{key} = ?")
params.append(value)
if not fields:
return {"status": "no_changes"}
params.append(expense_id)
conn = get_conn()
cur = conn.cursor()
cur.execute(f"UPDATE expenses SET {', '.join(fields)} WHERE id = ?", params)
conn.commit()
conn.close()
return {"status": "updated", "id": expense_id}
def delete_expense_from_db(expense_id: int) -> bool:
"""Delete an expense by ID."""
conn = get_conn()
cur = conn.cursor()
cur.execute("DELETE FROM expenses WHERE id = ?", (expense_id,))
conn.commit()
conn.close()
return True
# ============================================================
# READ / QUERY
# ============================================================
def list_expenses_from_db(category: str = "", start_date: str = "", end_date: str = "") -> List[Dict]:
"""Query expenses with optional filters."""
conn = get_conn()
cur = conn.cursor()
query = """
SELECT id, amount, category, subcategory, date, description
FROM expenses
WHERE 1=1
"""
params = []
if category:
query += " AND category = ?"
params.append(category)
if start_date:
query += " AND date >= ?"
params.append(parse_date(start_date))
if end_date:
query += " AND date <= ?"
params.append(parse_date(end_date))
query += " ORDER BY date DESC"
rows = cur.execute(query, params).fetchall()
conn.close()
return [_row_to_dict(r) for r in rows]
def get_expenses_by_month(month: str) -> List[Dict]:
"""Return all expenses within a month."""
start, end = get_month_range(month)
conn = get_conn()
cur = conn.cursor()
rows = cur.execute(
"""
SELECT id, amount, category, subcategory, date, description
FROM expenses
WHERE date BETWEEN ? AND ?
ORDER BY date ASC
""",
(start, end),
).fetchall()
conn.close()
return [_row_to_dict(r) for r in rows]
def summarize_expenses(start_date: str, end_date: str, category: Optional[str] = None) -> List[Dict]:
"""Summarize totals across categories."""
start = parse_date(start_date)
end = parse_date(end_date)
conn = get_conn()
cur = conn.cursor()
query = """
SELECT category, SUM(amount)
FROM expenses
WHERE date BETWEEN ? AND ?
"""
params = [start, end]
if category:
query += " AND category = ?"
params.append(category)
query += " GROUP BY category ORDER BY SUM(amount) DESC"
rows = cur.execute(query, params).fetchall()
conn.close()
return [{"category": r[0], "total": r[1]} for r in rows]
def get_expenses_by_month(month: str) -> List[Dict]:
"""
Return expenses for a given month (YYYY-MM or flexible month string).
Result: list of dicts with keys: id, amount, category, subcategory, date, description
"""
start_date, end_date = get_month_range(month)
conn = get_conn()
cur = conn.cursor()
rows = cur.execute(
"SELECT id, amount, category, subcategory, date, description "
"FROM expenses WHERE date BETWEEN ? AND ? ORDER BY date ASC",
(start_date, end_date)
).fetchall()
conn.close()
# Convert sqlite Row/tuple to dicts
result = []
for r in rows:
# handle sqlite3.Row or plain tuple
if hasattr(r, "keys"):
# sqlite Row: build dict from keys
result.append({k: r[k] for k in r.keys()})
else:
result.append({
"id": r[0],
"amount": r[1],
"category": r[2],
"subcategory": r[3],
"date": r[4],
"description": r[5],
})
return result
