31 lines
1.4 KiB
Python
31 lines
1.4 KiB
Python
from __future__ import annotations
|
|
|
|
from .shared import (
|
|
Any,
|
|
db,
|
|
fail,
|
|
new_id,
|
|
normalize_month,
|
|
now_iso,
|
|
ok,
|
|
row_to_dict,
|
|
)
|
|
|
|
|
|
def create_ledger_budget(month: str, category: str, amount: int | float, subcategory: str = "", person: str = "", currency: str = "KRW", memo: str = "", db_path: str | None = None) -> dict[str, Any]:
|
|
try:
|
|
month_norm = normalize_month(month); amount_int = int(amount)
|
|
except Exception as e:
|
|
return fail("validation_error", str(e))
|
|
if not category:
|
|
return fail("validation_error", "category is required")
|
|
if amount_int < 0:
|
|
return fail("validation_error", "amount must be non-negative")
|
|
ts = now_iso(); budget_id = new_id("bud")
|
|
with db(db_path) as conn:
|
|
conn.execute("""INSERT INTO budgets(id,month,category,subcategory,person,amount,currency,memo,created_at,updated_at)
|
|
VALUES(?,?,?,?,?,?,?,?,?,?)
|
|
ON CONFLICT(month,category,subcategory,person) DO UPDATE SET amount=excluded.amount,currency=excluded.currency,memo=excluded.memo,updated_at=excluded.updated_at""", (budget_id, month_norm, category, subcategory, person, amount_int, currency, memo, ts, ts))
|
|
row = conn.execute("SELECT * FROM budgets WHERE month=? AND category=? AND subcategory=? AND person=?", (month_norm, category, subcategory, person)).fetchone()
|
|
return ok(budget=row_to_dict(row))
|