Cheng Zhou
6cefa620e4
重新设计权限系统所有 UI 组件的视觉风格，统一配色、圆角、阴影和交互动效： - 实时概览：统计卡片加图标和渐变色条，健康评分改为环形进度，告警改为卡片式 - 趋势分析：图表卡片加彩色图标标识，ECharts 配色升级为渐变面积填充 - 访问日志：指标卡片带图标，日志审计改为卡片式入口，IP排行前三高亮 - IP属地：工具栏重设计，排行列表前三渐变高亮，指标卡片统一新风格 - 系统级权限：从 el-table 改为自定义卡片列表，模块块独立圆角卡片 - 项目权限配置：空状态引导优化，成员表格加头像，工具栏加背景容器 - 角色概览卡片：加进度条可视化，hover 微动效 - 接口权限矩阵：工具栏分离布局，表格圆角包裹 - 角色管理抽屉：侧边栏选中态渐变，操作行 hover 高亮 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
125 lines
3.7 KiB
Python
125 lines
3.7 KiB
Python
"""权限访问日志后台写入器
|
|
|
|
使用 asyncio.Queue 缓冲日志条目，批量写入数据库。
|
|
设计原则：
|
|
- enqueue() 非阻塞，不影响请求性能
|
|
- 队列满时丢弃（监控不应拖慢被监控系统）
|
|
- 批量写入减少数据库压力
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
from app.db.session import SessionLocal
|
|
from app.models.permission_access_log import PermissionAccessLog
|
|
|
|
logger = logging.getLogger("ctms.permission_log_writer")
|
|
|
|
BATCH_SIZE = 50
|
|
FLUSH_INTERVAL = 5.0
|
|
QUEUE_MAX_SIZE = 10000
|
|
|
|
|
|
class PermissionLogWriter:
|
|
|
|
def __init__(self):
|
|
self._queue: asyncio.Queue[dict] = asyncio.Queue(maxsize=QUEUE_MAX_SIZE)
|
|
self._task: asyncio.Task | None = None
|
|
|
|
def enqueue(self, entry: dict) -> None:
|
|
try:
|
|
self._queue.put_nowait(entry)
|
|
except asyncio.QueueFull:
|
|
logger.warning("Permission log queue full, dropping entry")
|
|
|
|
async def start(self) -> None:
|
|
self._task = asyncio.create_task(self._flush_loop())
|
|
logger.info("PermissionLogWriter started")
|
|
|
|
async def stop(self) -> None:
|
|
if self._task:
|
|
self._task.cancel()
|
|
try:
|
|
await self._task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
await self._drain()
|
|
logger.info("PermissionLogWriter stopped")
|
|
|
|
async def _flush_loop(self) -> None:
|
|
while True:
|
|
batch = await self._collect_batch()
|
|
if batch:
|
|
await self._write_batch(batch)
|
|
|
|
async def _collect_batch(self) -> list[dict]:
|
|
batch: list[dict] = []
|
|
try:
|
|
first = await asyncio.wait_for(self._queue.get(), timeout=FLUSH_INTERVAL)
|
|
batch.append(first)
|
|
except asyncio.TimeoutError:
|
|
return batch
|
|
|
|
while len(batch) < BATCH_SIZE:
|
|
try:
|
|
item = self._queue.get_nowait()
|
|
batch.append(item)
|
|
except asyncio.QueueEmpty:
|
|
break
|
|
return batch
|
|
|
|
async def _write_batch(self, batch: list[dict]) -> None:
|
|
try:
|
|
async with SessionLocal() as session:
|
|
for entry in batch:
|
|
log = PermissionAccessLog(
|
|
id=uuid.uuid4(),
|
|
study_id=entry["study_id"],
|
|
user_id=entry["user_id"],
|
|
endpoint_key=entry["endpoint_key"],
|
|
role=entry["role"],
|
|
allowed=entry["allowed"],
|
|
elapsed_ms=entry["elapsed_ms"],
|
|
ip_address=entry.get("ip_address"),
|
|
created_at=entry.get("created_at", datetime.now(timezone.utc)),
|
|
)
|
|
session.add(log)
|
|
await session.commit()
|
|
except Exception:
|
|
logger.exception("Failed to write permission access log batch (%d entries)", len(batch))
|
|
|
|
async def _drain(self) -> None:
|
|
batch: list[dict] = []
|
|
while not self._queue.empty():
|
|
try:
|
|
batch.append(self._queue.get_nowait())
|
|
except asyncio.QueueEmpty:
|
|
break
|
|
if batch:
|
|
await self._write_batch(batch)
|
|
|
|
|
|
_writer: PermissionLogWriter | None = None
|
|
|
|
|
|
def get_log_writer() -> PermissionLogWriter | None:
|
|
return _writer
|
|
|
|
|
|
async def start_log_writer() -> PermissionLogWriter:
|
|
global _writer
|
|
_writer = PermissionLogWriter()
|
|
await _writer.start()
|
|
return _writer
|
|
|
|
|
|
async def stop_log_writer() -> None:
|
|
global _writer
|
|
if _writer:
|
|
await _writer.stop()
|
|
_writer = None
|