Refactors transaction retrieval to use Fava API
Replaces direct database queries for transactions with calls to the Fava API, centralizing transaction logic and improving data consistency. This change removes redundant code and simplifies the API by relying on Fava for querying transactions based on account patterns and other criteria. Specifically, the commit introduces new methods in the FavaClient class for querying transactions, retrieving account transactions, and retrieving user transactions. The API endpoints are updated to utilize these methods.
This commit is contained in:
parent
88ff3821ce
commit
de3e4e65af
2 changed files with 208 additions and 106 deletions
108
fava_client.py
108
fava_client.py
|
|
@ -401,6 +401,114 @@ class FavaClient:
|
|||
logger.warning(f"Fava health check failed: {e}")
|
||||
return False
|
||||
|
||||
async def query_transactions(
|
||||
self,
|
||||
account_pattern: Optional[str] = None,
|
||||
limit: int = 100,
|
||||
include_pending: bool = True
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Query transactions from Fava/Beancount.
|
||||
|
||||
Args:
|
||||
account_pattern: Optional regex pattern to filter accounts (e.g., "User-abc123")
|
||||
limit: Maximum number of transactions to return
|
||||
include_pending: Include pending transactions (flag='!')
|
||||
|
||||
Returns:
|
||||
List of transaction dictionaries with date, description, postings, etc.
|
||||
|
||||
Example:
|
||||
# All transactions
|
||||
txns = await fava.query_transactions()
|
||||
|
||||
# User's transactions
|
||||
txns = await fava.query_transactions(account_pattern="User-abc123")
|
||||
|
||||
# Account transactions
|
||||
txns = await fava.query_transactions(account_pattern="Assets:Receivable:User-abc")
|
||||
"""
|
||||
# Build Beancount query
|
||||
if account_pattern:
|
||||
query = f"SELECT * WHERE account ~ '{account_pattern}' ORDER BY date DESC LIMIT {limit}"
|
||||
else:
|
||||
query = f"SELECT * ORDER BY date DESC LIMIT {limit}"
|
||||
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
||||
response = await client.get(
|
||||
f"{self.base_url}/query",
|
||||
params={"query_string": query}
|
||||
)
|
||||
response.raise_for_status()
|
||||
result = response.json()
|
||||
|
||||
# Fava query API returns: {"data": {"rows": [...], "columns": [...]}}
|
||||
data = result.get("data", {})
|
||||
rows = data.get("rows", [])
|
||||
|
||||
# Transform Fava's query result to transaction list
|
||||
transactions = []
|
||||
for row in rows:
|
||||
# Fava returns rows with various fields depending on the query
|
||||
# For "SELECT *", we get transaction details
|
||||
if isinstance(row, dict):
|
||||
# Filter by flag if needed
|
||||
flag = row.get("flag", "*")
|
||||
if not include_pending and flag == "!":
|
||||
continue
|
||||
|
||||
transactions.append(row)
|
||||
|
||||
return transactions[:limit]
|
||||
|
||||
except httpx.HTTPStatusError as e:
|
||||
logger.error(f"Fava query error: {e.response.status_code} - {e.response.text}")
|
||||
raise
|
||||
except httpx.RequestError as e:
|
||||
logger.error(f"Fava connection error: {e}")
|
||||
raise
|
||||
|
||||
async def get_account_transactions(
|
||||
self,
|
||||
account_name: str,
|
||||
limit: int = 100
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Get all transactions affecting a specific account.
|
||||
|
||||
Args:
|
||||
account_name: Full account name (e.g., "Assets:Receivable:User-abc123")
|
||||
limit: Maximum number of transactions
|
||||
|
||||
Returns:
|
||||
List of transactions affecting this account
|
||||
"""
|
||||
return await self.query_transactions(
|
||||
account_pattern=account_name.replace(":", "\\:"), # Escape colons for regex
|
||||
limit=limit
|
||||
)
|
||||
|
||||
async def get_user_transactions(
|
||||
self,
|
||||
user_id: str,
|
||||
limit: int = 100
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Get all transactions affecting a user's accounts.
|
||||
|
||||
Args:
|
||||
user_id: User ID
|
||||
limit: Maximum number of transactions
|
||||
|
||||
Returns:
|
||||
List of transactions affecting user's accounts
|
||||
"""
|
||||
return await self.query_transactions(
|
||||
account_pattern=f"User-{user_id[:8]}",
|
||||
limit=limit
|
||||
)
|
||||
|
||||
|
||||
# Singleton instance (configured from settings)
|
||||
_fava_client: Optional[FavaClient] = None
|
||||
|
|
|
|||
206
views_api.py
206
views_api.py
|
|
@ -26,14 +26,11 @@ from .crud import (
|
|||
get_account_by_name,
|
||||
get_account_permission,
|
||||
get_account_permissions,
|
||||
get_account_transactions,
|
||||
get_all_accounts,
|
||||
get_all_journal_entries,
|
||||
get_all_manual_payment_requests,
|
||||
get_all_user_wallet_settings,
|
||||
get_balance_assertion,
|
||||
get_balance_assertions,
|
||||
get_journal_entries_by_user,
|
||||
get_journal_entry,
|
||||
get_manual_payment_request,
|
||||
get_or_create_user_account,
|
||||
|
|
@ -252,24 +249,44 @@ async def api_get_account_balance(account_id: str) -> dict:
|
|||
|
||||
@castle_api_router.get("/api/v1/accounts/{account_id}/transactions")
|
||||
async def api_get_account_transactions(account_id: str, limit: int = 100) -> list[dict]:
|
||||
"""Get all transactions for an account"""
|
||||
transactions = await get_account_transactions(account_id, limit)
|
||||
return [
|
||||
{
|
||||
"journal_entry": entry.dict(),
|
||||
"entry_line": line.dict(),
|
||||
}
|
||||
for entry, line in transactions
|
||||
]
|
||||
"""
|
||||
Get all transactions for an account from Fava/Beancount.
|
||||
|
||||
Returns transactions affecting this account in reverse chronological order.
|
||||
"""
|
||||
from .fava_client import get_fava_client
|
||||
|
||||
# Get account details
|
||||
account = await get_account(account_id)
|
||||
if not account:
|
||||
raise HTTPException(
|
||||
status_code=HTTPStatus.NOT_FOUND,
|
||||
detail=f"Account {account_id} not found"
|
||||
)
|
||||
|
||||
# Query Fava for transactions
|
||||
fava = get_fava_client()
|
||||
transactions = await fava.get_account_transactions(account.name, limit)
|
||||
|
||||
return transactions
|
||||
|
||||
|
||||
# ===== JOURNAL ENTRY ENDPOINTS =====
|
||||
|
||||
|
||||
@castle_api_router.get("/api/v1/entries")
|
||||
async def api_get_journal_entries(limit: int = 100) -> list[JournalEntry]:
|
||||
"""Get all journal entries"""
|
||||
return await get_all_journal_entries(limit)
|
||||
async def api_get_journal_entries(limit: int = 100) -> list[dict]:
|
||||
"""
|
||||
Get all journal entries from Fava/Beancount.
|
||||
|
||||
Returns all transactions in reverse chronological order.
|
||||
"""
|
||||
from .fava_client import get_fava_client
|
||||
|
||||
fava = get_fava_client()
|
||||
transactions = await fava.query_transactions(limit=limit)
|
||||
|
||||
return transactions
|
||||
|
||||
|
||||
@castle_api_router.get("/api/v1/entries/user")
|
||||
|
|
@ -280,91 +297,56 @@ async def api_get_user_entries(
|
|||
filter_user_id: str = None,
|
||||
filter_account_type: str = None, # 'asset' for receivable, 'liability' for payable
|
||||
) -> dict:
|
||||
"""Get journal entries that affect the current user's accounts"""
|
||||
"""
|
||||
Get journal entries that affect the current user's accounts from Fava/Beancount.
|
||||
|
||||
Returns transactions in reverse chronological order with optional filtering.
|
||||
"""
|
||||
from lnbits.settings import settings as lnbits_settings
|
||||
from lnbits.core.crud.users import get_user
|
||||
from .crud import (
|
||||
count_all_journal_entries,
|
||||
count_journal_entries_by_user,
|
||||
count_journal_entries_by_user_and_account_type,
|
||||
get_account,
|
||||
get_journal_entries_by_user_and_account_type,
|
||||
)
|
||||
from .fava_client import get_fava_client
|
||||
|
||||
# Determine which entries to fetch based on filters
|
||||
fava = get_fava_client()
|
||||
|
||||
# Determine which user's entries to fetch
|
||||
if wallet.wallet.user == lnbits_settings.super_user:
|
||||
# Super user with user_id filter
|
||||
if filter_user_id:
|
||||
# Filter by both user_id and account_type
|
||||
if filter_account_type:
|
||||
entries = await get_journal_entries_by_user_and_account_type(
|
||||
filter_user_id, filter_account_type, limit, offset
|
||||
)
|
||||
total = await count_journal_entries_by_user_and_account_type(
|
||||
filter_user_id, filter_account_type
|
||||
)
|
||||
else:
|
||||
# Filter by user_id only
|
||||
entries = await get_journal_entries_by_user(filter_user_id, limit, offset)
|
||||
total = await count_journal_entries_by_user(filter_user_id)
|
||||
else:
|
||||
# No user filter, show all entries (account_type filter not supported for all entries)
|
||||
entries = await get_all_journal_entries(limit, offset)
|
||||
total = await count_all_journal_entries()
|
||||
# Super user can view all or filter by user_id
|
||||
target_user_id = filter_user_id
|
||||
else:
|
||||
# Regular user
|
||||
# Regular user can only see their own entries
|
||||
target_user_id = wallet.wallet.user
|
||||
|
||||
# Query Fava for transactions
|
||||
if target_user_id:
|
||||
# Build account pattern based on account_type filter
|
||||
if filter_account_type:
|
||||
entries = await get_journal_entries_by_user_and_account_type(
|
||||
wallet.wallet.user, filter_account_type, limit, offset
|
||||
)
|
||||
total = await count_journal_entries_by_user_and_account_type(
|
||||
wallet.wallet.user, filter_account_type
|
||||
)
|
||||
# Filter by account type (asset = receivable, liability = payable)
|
||||
if filter_account_type.lower() == "asset":
|
||||
account_pattern = f"Receivable:User-{target_user_id[:8]}"
|
||||
elif filter_account_type.lower() == "liability":
|
||||
account_pattern = f"Payable:User-{target_user_id[:8]}"
|
||||
else:
|
||||
account_pattern = f"User-{target_user_id[:8]}"
|
||||
else:
|
||||
entries = await get_journal_entries_by_user(wallet.wallet.user, limit, offset)
|
||||
total = await count_journal_entries_by_user(wallet.wallet.user)
|
||||
# All user accounts
|
||||
account_pattern = f"User-{target_user_id[:8]}"
|
||||
|
||||
# Enrich entries with username information
|
||||
enriched_entries = []
|
||||
for entry in entries:
|
||||
# Find user_id from entry lines (look for user-specific accounts)
|
||||
# Prioritize equity accounts, then liability/asset accounts
|
||||
entry_user_id = None
|
||||
entry_username = None
|
||||
entry_account_type = None
|
||||
|
||||
equity_account = None
|
||||
other_user_account = None
|
||||
|
||||
# First pass: look for equity and other user accounts
|
||||
for line in entry.lines:
|
||||
account = await get_account(line.account_id)
|
||||
if account and account.user_id:
|
||||
account_type = account.account_type.value if hasattr(account.account_type, 'value') else account.account_type
|
||||
|
||||
if account_type == 'equity':
|
||||
equity_account = (account.user_id, account_type, account)
|
||||
break # Prioritize equity, stop searching
|
||||
elif not other_user_account:
|
||||
other_user_account = (account.user_id, account_type, account)
|
||||
|
||||
# Use equity account if found, otherwise use other user account
|
||||
selected_account = equity_account or other_user_account
|
||||
|
||||
if selected_account:
|
||||
entry_user_id, entry_account_type, account_obj = selected_account
|
||||
user = await get_user(entry_user_id)
|
||||
entry_username = user.username if user and user.username else entry_user_id[:16] + "..."
|
||||
|
||||
enriched_entries.append({
|
||||
**entry.dict(),
|
||||
"user_id": entry_user_id,
|
||||
"username": entry_username,
|
||||
"account_type": entry_account_type,
|
||||
})
|
||||
entries = await fava.query_transactions(
|
||||
account_pattern=account_pattern,
|
||||
limit=limit + offset # Fava doesn't support offset, so fetch more and slice
|
||||
)
|
||||
# Apply offset
|
||||
entries = entries[offset:offset + limit]
|
||||
total = len(entries) # Note: This is approximate since we don't know the true total
|
||||
else:
|
||||
# Super user viewing all entries
|
||||
entries = await fava.query_transactions(limit=limit + offset)
|
||||
entries = entries[offset:offset + limit]
|
||||
total = len(entries)
|
||||
|
||||
# Fava transactions already contain the data we need
|
||||
# Metadata includes user-id, account information, etc.
|
||||
return {
|
||||
"entries": enriched_entries,
|
||||
"entries": entries,
|
||||
"total": total,
|
||||
"limit": limit,
|
||||
"offset": offset,
|
||||
|
|
@ -376,9 +358,14 @@ async def api_get_user_entries(
|
|||
@castle_api_router.get("/api/v1/entries/pending")
|
||||
async def api_get_pending_entries(
|
||||
wallet: WalletTypeInfo = Depends(require_admin_key),
|
||||
) -> list[JournalEntry]:
|
||||
"""Get all pending expense entries that need approval (admin only)"""
|
||||
) -> list[dict]:
|
||||
"""
|
||||
Get all pending expense entries that need approval (admin only).
|
||||
|
||||
Returns transactions with flag='!' from Fava/Beancount.
|
||||
"""
|
||||
from lnbits.settings import settings as lnbits_settings
|
||||
from .fava_client import get_fava_client
|
||||
|
||||
if wallet.wallet.user != lnbits_settings.super_user:
|
||||
raise HTTPException(
|
||||
|
|
@ -386,9 +373,12 @@ async def api_get_pending_entries(
|
|||
detail="Only super user can access this endpoint",
|
||||
)
|
||||
|
||||
# Get all journal entries and filter for pending flag
|
||||
all_entries = await get_all_journal_entries(limit=1000)
|
||||
pending_entries = [e for e in all_entries if e.flag == JournalEntryFlag.PENDING]
|
||||
# Query Fava for all transactions including pending
|
||||
fava = get_fava_client()
|
||||
all_entries = await fava.query_transactions(limit=1000, include_pending=True)
|
||||
|
||||
# Filter for pending flag
|
||||
pending_entries = [e for e in all_entries if e.get("flag") == "!"]
|
||||
return pending_entries
|
||||
|
||||
|
||||
|
|
@ -2141,14 +2131,16 @@ async def api_get_reconciliation_summary(
|
|||
failed = len([a for a in all_assertions if a.status == AssertionStatus.FAILED])
|
||||
pending = len([a for a in all_assertions if a.status == AssertionStatus.PENDING])
|
||||
|
||||
# Get all journal entries
|
||||
all_entries = await get_all_journal_entries(limit=1000)
|
||||
# Get all journal entries from Fava
|
||||
from .fava_client import get_fava_client
|
||||
fava = get_fava_client()
|
||||
all_entries = await fava.query_transactions(limit=1000, include_pending=True)
|
||||
|
||||
# Count entries by flag
|
||||
cleared = len([e for e in all_entries if e.flag == JournalEntryFlag.CLEARED])
|
||||
pending_entries = len([e for e in all_entries if e.flag == JournalEntryFlag.PENDING])
|
||||
flagged = len([e for e in all_entries if e.flag == JournalEntryFlag.FLAGGED])
|
||||
voided = len([e for e in all_entries if e.flag == JournalEntryFlag.VOID])
|
||||
cleared = len([e for e in all_entries if e.get("flag") == "*"])
|
||||
pending_entries = len([e for e in all_entries if e.get("flag") == "!"])
|
||||
flagged = len([e for e in all_entries if e.get("flag") == "#"])
|
||||
voided = len([e for e in all_entries if e.get("flag") == "x"])
|
||||
|
||||
# Get all accounts
|
||||
accounts = await get_all_accounts()
|
||||
|
|
@ -2231,10 +2223,12 @@ async def api_get_discrepancies(
|
|||
limit=1000,
|
||||
)
|
||||
|
||||
# Get flagged entries
|
||||
all_entries = await get_all_journal_entries(limit=1000)
|
||||
flagged_entries = [e for e in all_entries if e.flag == JournalEntryFlag.FLAGGED]
|
||||
pending_entries = [e for e in all_entries if e.flag == JournalEntryFlag.PENDING]
|
||||
# Get flagged entries from Fava
|
||||
from .fava_client import get_fava_client
|
||||
fava = get_fava_client()
|
||||
all_entries = await fava.query_transactions(limit=1000, include_pending=True)
|
||||
flagged_entries = [e for e in all_entries if e.get("flag") == "#"]
|
||||
pending_entries = [e for e in all_entries if e.get("flag") == "!"]
|
||||
|
||||
return {
|
||||
"failed_assertions": failed_assertions,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue