40c480e488
- Fix health check sqlExe calls: add missing ns parameter - Fix accounting ID generation: use getID() instead of uuid4 (VARCHAR length) - Fix accounting request_id: normalize empty to NULL to avoid UNIQUE constraint violation - Fix test_server balance/update route: was incorrectly pointing to accounting handler - Add test_server.py: standalone aiohttp test server for local development - Update conf/config.yaml: local MySQL credentials (test/test) - Update db/schema.sql and scripts/generate_ddl.py for local testing - Fix router sync_status sqlExe call: add missing ns parameter - Fix sync uapi_sync: use correct table/column names
130 lines
4.9 KiB
Python
130 lines
4.9 KiB
Python
"""
|
|
UAPISync - Sync uapi / upapp from Sage DB to uapi_cache.
|
|
|
|
Source tables (Sage DB):
|
|
- uapi
|
|
- upapp
|
|
|
|
Target table (cache DB):
|
|
- uapi_cache
|
|
"""
|
|
import logging
|
|
from typing import Dict, List, Optional
|
|
|
|
from .base_sync import BaseSync
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class UapiSync(BaseSync):
|
|
MODULE_NAME = "uapi"
|
|
SOURCE_DBNAME = "sage"
|
|
CACHE_DBNAME = "sageapi"
|
|
STATE_KEY = "sync_uapi"
|
|
BATCH_SIZE = 500
|
|
|
|
async def fetch_incremental(self, sor, since_timestamp: Optional[str]) -> List[Dict]:
|
|
"""
|
|
Fetch incremental data from uapi and upapp tables.
|
|
Joins API definitions with app registration data.
|
|
"""
|
|
if since_timestamp:
|
|
where_clause = f"WHERE u.updated_at > '{since_timestamp}' OR up.updated_at > '{since_timestamp}'"
|
|
else:
|
|
where_clause = ""
|
|
|
|
sql = f"""
|
|
SELECT
|
|
u.id AS uapi_id,
|
|
u.api_name,
|
|
u.api_path,
|
|
u.api_method,
|
|
u.api_version,
|
|
u.api_desc,
|
|
u.status AS uapi_status,
|
|
u.auth_required,
|
|
u.created_at AS uapi_created_at,
|
|
u.updated_at AS uapi_updated_at,
|
|
up.id AS upapp_id,
|
|
up.app_name,
|
|
up.app_code,
|
|
up.app_type,
|
|
up.app_desc,
|
|
up.app_owner,
|
|
up.status AS upapp_status,
|
|
up.updated_at AS upapp_updated_at
|
|
FROM {sor.dbname}.uapi u
|
|
LEFT JOIN {sor.dbname}.upapp up ON u.upapp_id = up.id
|
|
{where_clause}
|
|
ORDER BY COALESCE(u.updated_at, u.created_at) ASC
|
|
"""
|
|
|
|
records = await sor.sqlExe(sql, {})
|
|
return [dict(r) for r in records] if records else []
|
|
|
|
async def persist(self, sor, records: List[Dict]) -> None:
|
|
"""
|
|
UPSERT into uapi_cache using INSERT ... ON DUPLICATE KEY UPDATE.
|
|
The composite key is (uapi_id, upapp_id).
|
|
"""
|
|
import time
|
|
synced_at = str(int(time.time()))
|
|
|
|
for rec in records:
|
|
rec['synced_at'] = synced_at
|
|
insert_sql = """
|
|
INSERT INTO uapi_cache (
|
|
uapi_id, api_name, api_path, api_method, api_version,
|
|
api_desc, uapi_status, auth_required,
|
|
uapi_created_at, uapi_updated_at,
|
|
upapp_id, app_name, app_code, app_type,
|
|
app_desc, app_owner, upapp_status, upapp_updated_at,
|
|
synced_at
|
|
) VALUES (
|
|
${uapi_id}$, ${api_name}$, ${api_path}$, ${api_method}$, ${api_version}$,
|
|
${api_desc}$, ${uapi_status}$, ${auth_required}$,
|
|
${uapi_created_at}$, ${uapi_updated_at}$,
|
|
${upapp_id}$, ${app_name}$, ${app_code}$, ${app_type}$,
|
|
${app_desc}$, ${app_owner}$, ${upapp_status}$, ${upapp_updated_at}$,
|
|
${synced_at}$
|
|
)
|
|
ON DUPLICATE KEY UPDATE
|
|
api_name = VALUES(api_name),
|
|
api_path = VALUES(api_path),
|
|
api_method = VALUES(api_method),
|
|
api_version = VALUES(api_version),
|
|
api_desc = VALUES(api_desc),
|
|
uapi_status = VALUES(uapi_status),
|
|
auth_required = VALUES(auth_required),
|
|
uapi_created_at = VALUES(uapi_created_at),
|
|
uapi_updated_at = VALUES(uapi_updated_at),
|
|
app_name = VALUES(app_name),
|
|
app_code = VALUES(app_code),
|
|
app_type = VALUES(app_type),
|
|
app_desc = VALUES(app_desc),
|
|
app_owner = VALUES(app_owner),
|
|
upapp_status = VALUES(upapp_status),
|
|
upapp_updated_at = VALUES(upapp_updated_at),
|
|
synced_at = VALUES(synced_at)
|
|
"""
|
|
try:
|
|
await sor.execute(insert_sql, rec)
|
|
except Exception as e:
|
|
logger.warning(
|
|
"[%s] persist failed for uapi_id=%s: %s",
|
|
self.__class__.__name__, rec.get('uapi_id'), e,
|
|
)
|
|
raise
|
|
|
|
def get_latest_timestamp(self, records: List[Dict]) -> Optional[str]:
|
|
"""Extract the maximum updated_at from uapi or upapp records."""
|
|
if not records:
|
|
return None
|
|
latest = None
|
|
for r in records:
|
|
for key in ('uapi_updated_at', 'upapp_updated_at'):
|
|
ts = r.get(key)
|
|
if ts and (latest is None or str(ts) > str(latest)):
|
|
latest = str(ts)
|
|
return latest
|