jsfamily/jarca-ledger-mcp
5
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement BAT-011 transactional write service split

Browse Source
This commit is contained in:
Jarca
2026-07-03 07:42:16 +09:00
parent 7ac25a6cf7
commit ac78e94571
21 changed files with 844 additions and 228 deletions
Download Patch File Download Diff File Expand all files Collapse all files
jarca_ledger_mcp/application/system_services.py
+1 -16
View File
@@ -166,22 +166,7 @@ def get_ledger_backup_status(file_path: str, include_hash: bool = True, include_
return fail("not_found", f"backup file not found: {path}", path=str(path), exists=False)
return ok(backup=_ledger_db_status_for_path(path, include_hash=include_hash, include_integrity=include_integrity), restore_policy=BACKUP_RESTORE_POLICY)
def create_ledger_account(name: str, type: str = "bank", institution: str = "", currency: str = "KRW", active: bool = True, memo: str = "", account_number_masked: str = "", holder_name: str = "", last_balance: int | None = None, last_synced_at: str = "", statement_status: str = "unknown", liquidity_bucket: str = "other", db_path: str | None = None) -> dict[str, Any]:
if not name.strip():
return fail("validation_error", "account name is required")
if type not in VALID_ACCOUNT_TYPES:
return fail("validation_error", f"type must be one of {sorted(VALID_ACCOUNT_TYPES)}")
try:
statement_status_norm = normalize_statement_status(statement_status)
liquidity_bucket_norm = normalize_liquidity_bucket(liquidity_bucket)
except Exception as e:
return fail("validation_error", str(e))
ts = now_iso(); account_id = new_id("acct")
with db(db_path) as conn:
conn.execute("""INSERT INTO accounts(id,name,type,institution,currency,active,memo,account_number_masked,holder_name,last_balance,last_synced_at,statement_status,liquidity_bucket,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", (account_id, name.strip(), type, institution, currency, int(active), memo, account_number_masked, holder_name, last_balance, last_synced_at, statement_status_norm, liquidity_bucket_norm, ts, ts))
row = conn.execute("SELECT * FROM accounts WHERE id=?", (account_id,)).fetchone()
return ok(account=row_to_dict(row))
from .accounts_write_services import create_ledger_account
def list_ledger_accounts(active_only: bool = True, db_path: str | None = None) -> dict[str, Any]:
sql = "SELECT * FROM accounts" + (" WHERE active=1" if active_only else "") + " ORDER BY name"