5936a2f328
- Sync engine: BaseSync abstract class + 4 sync modules (users/pricing/uapi/llmage) - Checkpoint management via sync_state table - Batch processing with retry and exponential backoff - Incremental fetch from Sage DB via sqlor - UPSERT to local cache tables - API handlers: balance/accounting/users/pricing/health - Balance: cache lookup + Sage fallback - Accounting: create with idempotency, query with filters/pagination - Users: keyword search, org filter - Pricing: filter by ppid/llmid/type/status - Health: basic + readiness checks (DB connectivity) - DAPI auth: middleware + authenticate_request function - HMAC-SHA256 signature verification - Timestamp window validation - Sage downapikey table lookup - HTTP client: SageHttpClient with aiohttp - Auto DAPI signature injection - Connection pooling, retry, timeout - Router: 12 routes registered - Module init: load_sageapi() wires everything to ServerEnv
122 lines
4.1 KiB
Python
122 lines
4.1 KiB
Python
"""Customer balance query API handler.
|
|
|
|
Provides the RESTful endpoint for querying customer account balances.
|
|
Reads from the local customer_balance cache table, with fallback to
|
|
real-time query from Sage acc_balance table.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
from appPublic.log import debug, error
|
|
from sqlor.dbpools import DBPools, get_sor_context
|
|
from ahserver.serverenv import ServerEnv
|
|
|
|
|
|
async def get_customer_balance(customer_id: str | None = None) -> str:
|
|
"""Query customer balance.
|
|
|
|
First checks the local customer_balance cache. If not found,
|
|
falls back to real-time query from Sage acc_balance table.
|
|
|
|
Args:
|
|
customer_id: Optional customer ID filter.
|
|
|
|
Returns:
|
|
JSON string with success flag and balance data.
|
|
"""
|
|
result: dict[str, Any] = {'success': False, 'data': [], 'total': 0}
|
|
|
|
try:
|
|
env = ServerEnv()
|
|
cache_dbname = env.get_module_dbname('sageapi')
|
|
if not cache_dbname:
|
|
result['error'] = 'No database configured for sageapi module'
|
|
return json.dumps(result, ensure_ascii=False, default=str)
|
|
|
|
async with DBPools().sqlorContext(cache_dbname) as sor:
|
|
params: dict[str, Any] = {}
|
|
where = ''
|
|
if customer_id:
|
|
where = 'WHERE id = ${customer_id}$'
|
|
params['customer_id'] = customer_id
|
|
|
|
sql = f"""
|
|
SELECT id, balance, currency, credit_limit,
|
|
last_recharge, last_consumption,
|
|
status, cached_at
|
|
FROM customer_balance
|
|
{where}
|
|
ORDER BY id
|
|
"""
|
|
data = await sor.sqlExe(sql, params)
|
|
if isinstance(data, dict):
|
|
result['total'] = data.get('total', len(data.get('rows', [])))
|
|
result['data'] = data.get('rows', [])
|
|
elif isinstance(data, list):
|
|
result['total'] = len(data)
|
|
result['data'] = data
|
|
|
|
# If cache miss for specific customer, try real-time from Sage
|
|
if customer_id and not result['data']:
|
|
result['data'] = await _query_sage_balance(env, customer_id)
|
|
result['total'] = len(result['data'])
|
|
|
|
result['success'] = True
|
|
|
|
except Exception as e:
|
|
error(f'get_customer_balance error: {e}')
|
|
result['error'] = str(e)
|
|
|
|
return json.dumps(result, ensure_ascii=False, default=str)
|
|
|
|
|
|
async def _query_sage_balance(env: ServerEnv, customer_id: str) -> list[dict]:
|
|
"""Fallback: query Sage acc_balance table directly."""
|
|
try:
|
|
async with get_sor_context(env, 'sage') as sor:
|
|
sql = """
|
|
SELECT customer_id, balance, currency, status, updated_at
|
|
FROM acc_balance
|
|
WHERE customer_id = ${customer_id}$
|
|
"""
|
|
rows = await sor.sqlExe(sql, {'customer_id': customer_id})
|
|
if isinstance(rows, list):
|
|
return rows
|
|
elif isinstance(rows, dict):
|
|
return rows.get('rows', [])
|
|
except Exception as e:
|
|
error(f'_query_sage_balance error: {e}')
|
|
return []
|
|
|
|
|
|
async def update_customer_balance(customer_id: str, balance: float) -> str:
|
|
"""Update customer balance in cache (called by sync or accounting)."""
|
|
result: dict[str, Any] = {'success': False}
|
|
|
|
try:
|
|
env = ServerEnv()
|
|
cache_dbname = env.get_module_dbname('sageapi')
|
|
|
|
sql = """
|
|
INSERT INTO customer_balance (id, balance, cached_at)
|
|
VALUES (${customer_id}$, ${balance}$, NOW())
|
|
ON DUPLICATE KEY UPDATE
|
|
balance = ${balance}$,
|
|
cached_at = NOW()
|
|
"""
|
|
async with DBPools().sqlorContext(cache_dbname) as sor:
|
|
await sor.sqlExe(sql, {
|
|
'customer_id': customer_id,
|
|
'balance': balance,
|
|
})
|
|
result['success'] = True
|
|
|
|
except Exception as e:
|
|
error(f'update_customer_balance error: {e}')
|
|
result['error'] = str(e)
|
|
|
|
return json.dumps(result, ensure_ascii=False, default=str)
|