31c9117ae7
All checks were successful
E2E Health Check / e2e-health (push) Successful in 24s
修復內容: 1. e2e_network_test.py: 移除 unittest.mock - 將 16 個 patch.object 改為 pytest monkeypatch - 符合 feedback_no_mock_testing.md 2. audit_logs.py: Router→Service 層封裝 - 新增 AuditLogService (audit_log_service.py) - Router 改用 get_audit_log_service() - 移除直接 Repository 存取 3. incidents.py:463: DEBUG 端點重構 - 移除 get_incident_repository() 直接呼叫 - 完全透過 IncidentService 操作 - 簡化回傳結構 遵循規範: - Skill 09: Router 層禁止直接外部 API 呼叫 - feedback_lewooogo_modular_enforcement.md: Service 層封裝 - feedback_no_mock_testing.md: 禁止 MagicMock/AsyncMock Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
143 lines
3.5 KiB
Python
143 lines
3.5 KiB
Python
"""
|
||
Audit Log Service - Phase 22 P1 模組化修復
|
||
==========================================
|
||
Router 層禁止直接存取 Repository，抽取至 Service 層
|
||
|
||
遵循規範:
|
||
- Skill 09: Router 層禁止直接外部 API 呼叫
|
||
- feedback_lewooogo_modular_enforcement.md: Service 層封裝
|
||
|
||
功能:
|
||
- 分頁查詢稽核日誌
|
||
- 取得單筆稽核日誌
|
||
- 統計資訊查詢
|
||
|
||
版本: v1.0
|
||
建立: 2026-03-31 (台北時區)
|
||
建立者: Claude Code (首席架構師 P1 修復)
|
||
"""
|
||
|
||
from typing import Any
|
||
|
||
import structlog
|
||
|
||
from src.db.models import AuditLog
|
||
from src.repositories.audit_log_repository import get_audit_log_repository
|
||
|
||
logger = structlog.get_logger(__name__)
|
||
|
||
|
||
# =============================================================================
|
||
# Audit Log Service
|
||
# =============================================================================
|
||
|
||
|
||
class AuditLogService:
|
||
"""
|
||
Audit Log Service
|
||
|
||
統一稽核日誌存取，符合 leWOOOgo 積木化原則
|
||
|
||
2026-03-31 Claude Code (Phase 22 P1 修復)
|
||
"""
|
||
|
||
def __init__(self) -> None:
|
||
self._repository = get_audit_log_repository()
|
||
|
||
async def list_logs(
|
||
self,
|
||
page: int = 1,
|
||
page_size: int = 20,
|
||
success_only: bool | None = None,
|
||
namespace: str | None = None,
|
||
operation_type: str | None = None,
|
||
) -> tuple[list[AuditLog], int]:
|
||
"""
|
||
分頁查詢稽核日誌
|
||
|
||
Args:
|
||
page: 頁碼 (從 1 開始)
|
||
page_size: 每頁筆數
|
||
success_only: 篩選成功/失敗
|
||
namespace: 篩選 Namespace
|
||
operation_type: 篩選操作類型 (暫未實作)
|
||
|
||
Returns:
|
||
(logs, total_count)
|
||
"""
|
||
logs, total_count = await self._repository.list_logs(
|
||
page=page,
|
||
page_size=page_size,
|
||
success_only=success_only,
|
||
namespace=namespace,
|
||
)
|
||
|
||
logger.debug(
|
||
"audit_logs_listed",
|
||
page=page,
|
||
count=len(logs),
|
||
total=total_count,
|
||
)
|
||
|
||
return logs, total_count
|
||
|
||
async def get_by_id(self, log_id: str) -> AuditLog | None:
|
||
"""
|
||
取得單筆稽核日誌
|
||
|
||
Args:
|
||
log_id: 稽核日誌 ID
|
||
|
||
Returns:
|
||
AuditLog | None
|
||
"""
|
||
log = await self._repository.get_by_id(log_id)
|
||
|
||
if log:
|
||
logger.debug(
|
||
"audit_log_fetched",
|
||
log_id=log_id,
|
||
)
|
||
else:
|
||
logger.debug(
|
||
"audit_log_not_found",
|
||
log_id=log_id,
|
||
)
|
||
|
||
return log
|
||
|
||
async def get_stats(self, since_hours: int = 24) -> dict[str, Any]:
|
||
"""
|
||
取得稽核統計資訊
|
||
|
||
Args:
|
||
since_hours: 統計時間範圍 (小時)
|
||
|
||
Returns:
|
||
統計資訊 dict
|
||
"""
|
||
stats = await self._repository.get_stats(since_hours=since_hours)
|
||
|
||
logger.debug(
|
||
"audit_stats_fetched",
|
||
total=stats.get("total", 0),
|
||
success_rate=stats.get("success_rate", 0.0),
|
||
)
|
||
|
||
return stats
|
||
|
||
|
||
# =============================================================================
|
||
# Singleton
|
||
# =============================================================================
|
||
|
||
_audit_log_service: AuditLogService | None = None
|
||
|
||
|
||
def get_audit_log_service() -> AuditLogService:
|
||
"""取得 AuditLogService singleton"""
|
||
global _audit_log_service
|
||
if _audit_log_service is None:
|
||
_audit_log_service = AuditLogService()
|
||
return _audit_log_service
|