ForcePilot/backend/package/yuxi/external_systems/adapters/persistence/mappers.py
Kris 830a3c1458 refactor: 完成多模块代码优化与功能迭代
本次提交包含多项核心改进:
1. 统一列表查询分页范式,替换page/page_size为limit/offset标准分页
2. 新增多项仓库方法与服务接口,完善统计与测试能力
3. 优化审计日志事务处理逻辑,保证业务与审计数据原子性
4. 统一适配器类型与源类型排序契约,添加排序注释说明
5. 完善通知渠道、健康检查、告警等模块的DTO与实体字段
6. 优化测试用例执行回调,透传调用方标识
7. 修复资产导入时的校验逻辑,重新计算size与checksum
8. 拆分UpsertQuotaInput的审计字段,区分创建人与修改人
2026-07-11 06:55:19 +08:00

607 lines
22 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""ORM ↔ dataclass 映射与加解密边界。
本模块是 ``adapters/persistence`` 层**唯一**调用 ``yuxi.utils.crypto`` 加解密函数的位置。
所有 ORM 实例 ↔ dataclass 转换在此完成core 层不接触 ORMdataclass 始终持有明文。
设计原则(见设计方案 §4
- **mapper 不调用 ORM 的 ``to_dict`` 方法**:直接读 ORM 字段构造 dataclass。
原因:``ExternalSystemToken.to_dict`` / ``ExternalWebhookSubscription.to_dict``
不返回敏感字段明文(仅返回 ``has_access_token`` / ``has_secret`` 布尔字段),
且 ``to_dict`` 返回的 dict 结构与 dataclass 字段不完全一致。
- **5 个含敏感字段的 dataclass** 在读取时解密、写入前加密:
- dict 型敏感字段(``auth_config`` / ``secret_refs`` / ``adapter_config``)使用
``encrypt_sensitive_fields`` / ``decrypt_sensitive_fields``
- 单值敏感字段(``access_token`` / ``refresh_token`` / ``secret``)使用
``encrypt`` / ``decrypt``。
- **13 个无敏感字段的 dataclass** 直接字段映射。
- **解密失败时向上抛 ``ValueError``**,不掩盖密钥变更问题。
- ``ExternalSystem.env_count`` 是 ORM ``column_property`` 动态计算字段mapper
读取时直接读 ``orm.env_count``dataclass → ORM 转换时跳过该字段(不写入 DB
"""
from __future__ import annotations
from typing import Any
from yuxi.external_systems.core.models import (
AdapterAsset,
Alert,
AuditLog,
Environment,
Execution,
ExecutionTrace,
ExternalSystem,
ExternalTool,
HealthCheck,
Metric,
NotificationChannel,
QuotaUsage,
SecretRotationPolicy,
SystemToken,
ToolAccessRule,
ToolTestCase,
ToolVersion,
WebhookEvent,
WebhookSubscription,
)
from yuxi.storage.postgres.models_external import (
ExternalAdapterAsset,
ExternalNotificationChannel,
ExternalSecretRotationPolicy,
ExternalSystem as ExternalSystemORM,
ExternalSystemAlert,
ExternalSystemAuditLog,
ExternalSystemEnvironment,
ExternalSystemHealthCheck,
ExternalSystemMetric,
ExternalSystemQuotaUsage,
ExternalSystemToken,
ExternalToolAccessRule,
ExternalToolConfig,
ExternalToolConfigVersion,
ExternalToolExecution,
ExternalToolExecutionTrace,
ExternalToolTestCase,
ExternalWebhookEvent,
ExternalWebhookSubscription,
)
from yuxi.utils.crypto import (
decrypt,
decrypt_sensitive_fields,
encrypt,
encrypt_sensitive_fields,
)
# ═══════════════════════════════════════════════════════════════════════════
# 含敏感字段的 dataclass读取时解密 / 写入前加密)
# ═══════════════════════════════════════════════════════════════════════════
# ─── ExternalSystemdict 型敏感字段auth_config / secret_refs ──────────
def orm_to_external_system(orm: ExternalSystemORM) -> ExternalSystem:
"""ORM → dataclass解密 ``auth_config`` / ``secret_refs``。
解密失败时抛 ``ValueError``(密钥变更场景),由调用方决定处理策略。
``env_count`` 为 ORM ``column_property`` 动态计算字段,直接读取。
"""
return ExternalSystem(
id=orm.id,
slug=orm.slug,
name=orm.name,
description=orm.description,
category=orm.category,
adapter_type=orm.adapter_type,
source_type=orm.source_type,
enabled=orm.enabled,
connection_config=orm.connection_config or {},
auth_type=orm.auth_type,
auth_config=decrypt_sensitive_fields(orm.auth_config) or {},
secret_refs=decrypt_sensitive_fields(orm.secret_refs) or {},
rate_limit=orm.rate_limit or {},
circuit_breaker=orm.circuit_breaker or {},
pool_config=orm.pool_config or {},
observability=orm.observability or {},
timeout=orm.timeout,
retry_policy=orm.retry_policy or {},
created_by=orm.created_by,
updated_by=orm.updated_by,
env_count=orm.env_count,
)
def encrypt_system_for_write(data: dict[str, Any]) -> dict[str, Any]:
"""明文 dict → 密文 dict加密 ``auth_config`` / ``secret_refs``。
注意:``encrypt`` 不幂等,调用方必须保证传入的是明文(未被加密过)。
"""
data = dict(data)
if data.get("auth_config") is not None:
data["auth_config"] = encrypt_sensitive_fields(data["auth_config"])
if data.get("secret_refs") is not None:
data["secret_refs"] = encrypt_sensitive_fields(data["secret_refs"])
return data
# ─── Environmentdict 型敏感字段auth_config ──────────────────────────
def orm_to_environment(orm: ExternalSystemEnvironment) -> Environment:
"""ORM → dataclass解密 ``auth_config``。"""
return Environment(
id=orm.id,
system_id=orm.system_id,
env_key=orm.env_key,
name=orm.name,
connection_config=orm.connection_config or {},
auth_config=decrypt_sensitive_fields(orm.auth_config) or {},
is_default=orm.is_default,
enabled=orm.enabled,
created_by=orm.created_by,
updated_by=orm.updated_by,
created_at=orm.created_at,
updated_at=orm.updated_at,
)
def encrypt_environment_for_write(data: dict[str, Any]) -> dict[str, Any]:
"""明文 dict → 密文 dict加密 ``auth_config``。"""
data = dict(data)
if data.get("auth_config") is not None:
data["auth_config"] = encrypt_sensitive_fields(data["auth_config"])
return data
# ─── ExternalTooldict 型敏感字段auth_config / adapter_config ──────────
def orm_to_external_tool(orm: ExternalToolConfig) -> ExternalTool:
"""ORM → dataclass解密 ``auth_config`` / ``adapter_config``。
注意:``adapter_config`` 在 ORM 中 ``nullable=False``,但可能包含敏感字段
(如 mTLS 证书路径),统一走 ``decrypt_sensitive_fields``。
"""
return ExternalTool(
id=orm.id,
slug=orm.slug,
name=orm.name,
description=orm.description,
category=orm.category,
adapter_type=orm.adapter_type,
auth_type=orm.auth_type,
adapter_config=decrypt_sensitive_fields(orm.adapter_config) or {},
auth_config=decrypt_sensitive_fields(orm.auth_config) or {},
timeout=orm.timeout,
retry_policy=orm.retry_policy or {},
enabled=orm.enabled,
system_id=orm.system_id,
created_by=orm.created_by,
updated_by=orm.updated_by,
)
def encrypt_tool_for_write(data: dict[str, Any]) -> dict[str, Any]:
"""明文 dict → 密文 dict加密 ``auth_config`` / ``adapter_config``。"""
data = dict(data)
if data.get("auth_config") is not None:
data["auth_config"] = encrypt_sensitive_fields(data["auth_config"])
if data.get("adapter_config") is not None:
data["adapter_config"] = encrypt_sensitive_fields(data["adapter_config"])
return data
# ─── SystemToken单值加密access_token / refresh_token ─────────────────
def orm_to_system_token(orm: ExternalSystemToken) -> SystemToken:
"""ORM → dataclass单值解密 ``access_token`` / ``refresh_token``。
注意:不调用 ``orm.to_dict()``(它只返回 ``has_access_token`` 布尔字段)。
"""
return SystemToken(
id=orm.id,
system_id=orm.system_id,
env_key=orm.env_key,
token_type=orm.token_type,
access_token=decrypt(orm.access_token) if orm.access_token else "",
refresh_token=decrypt(orm.refresh_token) if orm.refresh_token else None,
expires_at=orm.expires_at,
extra=orm.extra or {},
is_invalidated=orm.is_invalidated,
)
def encrypt_token_for_write(data: dict[str, Any]) -> dict[str, Any]:
"""明文 dict → 密文 dict单值加密 ``access_token`` / ``refresh_token``。"""
data = dict(data)
if data.get("access_token"):
data["access_token"] = encrypt(data["access_token"])
if data.get("refresh_token"):
data["refresh_token"] = encrypt(data["refresh_token"])
return data
# ─── WebhookSubscription单值加密secret ──────────────────────────────
def orm_to_webhook_subscription(orm: ExternalWebhookSubscription) -> WebhookSubscription:
"""ORM → dataclass单值解密 ``secret``。
注意:不调用 ``orm.to_dict()``(它只返回 ``has_secret`` 布尔字段)。
"""
return WebhookSubscription(
id=orm.id,
slug=orm.slug,
name=orm.name,
system_id=orm.system_id,
source_event_type=orm.source_event_type,
target_handler_type=orm.target_handler_type,
callback_path=orm.callback_path,
description=orm.description or "",
env_key=orm.env_key,
subscription_id=orm.subscription_id,
target_handler_id=orm.target_handler_id,
secret=decrypt(orm.secret) if orm.secret else None,
secret_algorithm=orm.secret_algorithm,
filter_rules=orm.filter_rules or {},
transform_rules=orm.transform_rules or {},
status=orm.status,
enabled=orm.enabled,
expires_at=orm.expires_at,
auto_renew=orm.auto_renew,
last_renewed_at=orm.last_renewed_at,
last_renew_error=orm.last_renew_error,
renewal_attempts=orm.renewal_attempts,
last_received_at=orm.last_received_at,
received_count=orm.received_count,
retention_days=orm.retention_days,
created_by=orm.created_by,
updated_by=orm.updated_by,
)
def encrypt_webhook_subscription_for_write(data: dict[str, Any]) -> dict[str, Any]:
"""明文 dict → 密文 dict单值加密 ``secret``。"""
data = dict(data)
if data.get("secret"):
data["secret"] = encrypt(data["secret"])
return data
# ═══════════════════════════════════════════════════════════════════════════
# 无敏感字段的 dataclass直接字段映射
# ═══════════════════════════════════════════════════════════════════════════
def orm_to_tool_version(orm: ExternalToolConfigVersion) -> ToolVersion:
"""ORM → dataclass无敏感字段直接映射。"""
return ToolVersion(
id=orm.id,
tool_slug=orm.tool_slug,
version_id=orm.version_id,
snapshot=orm.snapshot or {},
created_by=orm.created_by,
created_at=orm.created_at,
)
def orm_to_adapter_asset(orm: ExternalAdapterAsset) -> AdapterAsset:
"""ORM → dataclass无敏感字段直接映射。``content`` 为二进制原值。"""
return AdapterAsset(
id=orm.id,
adapter_type=orm.adapter_type,
asset_type=orm.asset_type,
name=orm.name,
content=orm.content or b"",
size=orm.size or 0,
checksum=orm.checksum,
status=orm.status,
status_message=orm.status_message,
created_by=orm.created_by,
updated_by=orm.updated_by,
created_at=orm.created_at,
updated_at=orm.updated_at,
)
def orm_to_execution(orm: ExternalToolExecution) -> Execution:
"""ORM → dataclass无敏感字段直接映射。"""
return Execution(
id=orm.id,
execution_id=orm.execution_id,
trace_id=orm.trace_id,
system_id=orm.system_id,
env_key=orm.env_key,
tool_slug=orm.tool_slug,
caller=orm.caller,
caller_id=orm.caller_id,
status=orm.status,
started_at=orm.started_at,
operation=orm.operation,
request_summary=orm.request_summary or {},
response_summary=orm.response_summary,
error=orm.error,
ended_at=orm.ended_at,
duration_ms=orm.duration_ms,
correlation_id=orm.correlation_id,
retry_of=orm.retry_of,
retry_count=orm.retry_count,
tags=orm.tags or {},
)
def orm_to_execution_trace(orm: ExternalToolExecutionTrace) -> ExecutionTrace:
"""ORM → dataclass无敏感字段直接映射。"""
return ExecutionTrace(
id=orm.id,
execution_id=orm.execution_id,
trace_id=orm.trace_id,
spans=orm.spans or {},
total_duration_ms=orm.total_duration_ms,
created_by=orm.created_by,
updated_by=orm.updated_by,
created_at=orm.created_at,
updated_at=orm.updated_at,
)
def orm_to_audit_log(orm: ExternalSystemAuditLog) -> AuditLog:
"""ORM → dataclass无敏感字段直接映射。"""
return AuditLog(
id=orm.id,
system_id=orm.system_id,
env_key=orm.env_key,
event_type=orm.event_type,
action_type=orm.action_type,
resource=orm.resource,
user=orm.user,
detail=orm.detail or {},
snapshot_before=orm.snapshot_before,
snapshot_after=orm.snapshot_after,
description=orm.description,
created_by=orm.created_by,
updated_by=orm.updated_by,
created_at=orm.created_at,
)
def orm_to_metric(orm: ExternalSystemMetric) -> Metric:
"""ORM → dataclass无敏感字段直接映射。"""
return Metric(
id=orm.id,
system_id=orm.system_id,
env_key=orm.env_key,
bucket_at=orm.bucket_at,
total_calls=orm.total_calls,
success_calls=orm.success_calls,
failed_calls=orm.failed_calls,
auth_failed_calls=orm.auth_failed_calls,
ssrf_blocked_calls=orm.ssrf_blocked_calls,
timeout_calls=orm.timeout_calls,
latency_sum_ms=orm.latency_sum_ms,
latency_count=orm.latency_count,
p99_latency_ms=orm.p99_latency_ms,
token_refresh_count=orm.token_refresh_count,
token_refresh_failed=orm.token_refresh_failed,
throttled_count=orm.throttled_count,
circuit_open_count=orm.circuit_open_count,
)
def orm_to_health_check(orm: ExternalSystemHealthCheck) -> HealthCheck:
"""ORM → dataclass无敏感字段直接映射。"""
return HealthCheck(
id=orm.id,
system_id=orm.system_id,
env_key=orm.env_key,
health_status=orm.health_status,
duration_ms=orm.duration_ms,
detail=orm.detail,
error_code=orm.error_code,
error_message=orm.error_message,
checked_at=orm.checked_at,
)
def orm_to_webhook_event(orm: ExternalWebhookEvent) -> WebhookEvent:
"""ORM → dataclass无敏感字段直接映射。"""
return WebhookEvent(
id=orm.id,
subscription_id=orm.subscription_id,
subscription_slug=orm.subscription_slug,
event_type=orm.event_type,
payload=orm.payload or {},
received_at=orm.received_at,
external_event_id=orm.external_event_id,
payload_hash=orm.payload_hash,
headers=orm.headers or {},
signature=orm.signature,
signature_verified=orm.signature_verified,
source_ip=orm.source_ip,
processing_status=orm.processing_status,
processing_attempts=orm.processing_attempts,
last_processed_at=orm.last_processed_at,
processing_error=orm.processing_error,
correlation_id=orm.correlation_id,
triggered_handler_type=orm.triggered_handler_type,
triggered_handler_id=orm.triggered_handler_id,
triggered_execution_id=orm.triggered_execution_id,
)
def orm_to_quota_usage(orm: ExternalSystemQuotaUsage) -> QuotaUsage:
"""ORM → dataclass无敏感字段直接映射。"""
return QuotaUsage(
id=orm.id,
system_id=orm.system_id,
quota_key=orm.quota_key,
quota_name=orm.quota_name,
quota_window=orm.quota_window,
env_key=orm.env_key,
window_start=orm.window_start,
window_end=orm.window_end,
limit_value=orm.limit_value,
used_value=orm.used_value,
unit=orm.unit,
source=orm.source,
source_header_name=orm.source_header_name,
source_header_strategy=orm.source_header_strategy,
warning_threshold=orm.warning_threshold,
critical_threshold=orm.critical_threshold,
last_updated_at=orm.last_updated_at,
last_response_at=orm.last_response_at,
)
def orm_to_alert(orm: ExternalSystemAlert) -> Alert:
"""ORM → dataclass无敏感字段直接映射。
``system_name`` / ``system_slug`` 非 ORM 列,由 ``SqlAlchemyAlertRepository.list``
在 JOIN 查询后通过 ``dataclasses.replace`` 填充。
"""
return Alert(
id=orm.id,
system_id=orm.system_id,
alert_type=orm.alert_type,
title=orm.title,
triggered_at=orm.triggered_at,
env_key=orm.env_key,
severity=orm.severity,
status=orm.status,
description=orm.description,
detail=orm.detail or {},
resource_type=orm.resource_type,
resource_id=orm.resource_id,
related_execution_id=orm.related_execution_id,
related_trace_id=orm.related_trace_id,
metric_snapshot=orm.metric_snapshot,
dedup_key=orm.dedup_key,
resolved_at=orm.resolved_at,
acknowledged_by=orm.acknowledged_by,
acknowledged_at=orm.acknowledged_at,
resolution_note=orm.resolution_note,
notification_status=orm.notification_status,
notification_sent_at=orm.notification_sent_at,
notification_channels=orm.notification_channels or {},
created_by=orm.created_by,
updated_by=orm.updated_by,
created_at=orm.created_at,
updated_at=orm.updated_at,
)
def orm_to_secret_rotation_policy(orm: ExternalSecretRotationPolicy) -> SecretRotationPolicy:
"""ORM → dataclass无敏感字段直接映射。
注意:``secret_ref`` 仅是引用 key指向 ``ExternalSystem.secret_refs`` 中的具名 secret
非敏感值本身,不加密。
"""
return SecretRotationPolicy(
id=orm.id,
system_id=orm.system_id,
secret_type=orm.secret_type,
secret_name=orm.secret_name,
secret_ref=orm.secret_ref,
env_key=orm.env_key,
rotation_mode=orm.rotation_mode,
rotation_interval_days=orm.rotation_interval_days,
notify_before_days=orm.notify_before_days,
notify_channels=orm.notify_channels or {},
last_rotated_at=orm.last_rotated_at,
next_rotation_at=orm.next_rotation_at,
last_rotation_at=orm.last_rotation_at,
last_rotation_status=orm.last_rotation_status,
last_rotation_error=orm.last_rotation_error,
rotation_count=orm.rotation_count,
max_rotation_retries=orm.max_rotation_retries,
current_retry_count=orm.current_retry_count,
status=orm.status,
)
def orm_to_tool_access_rule(orm: ExternalToolAccessRule) -> ToolAccessRule:
"""ORM → dataclass无敏感字段直接映射。"""
return ToolAccessRule(
id=orm.id,
principal_type=orm.principal_type,
principal_id=orm.principal_id,
created_by=orm.created_by,
tool_slug=orm.tool_slug,
system_id=orm.system_id,
principal_name=orm.principal_name,
env_key=orm.env_key,
effect=orm.effect,
priority=orm.priority,
conditions=orm.conditions or {},
expires_at=orm.expires_at,
enabled=orm.enabled,
description=orm.description,
updated_by=orm.updated_by,
created_at=orm.created_at,
updated_at=orm.updated_at,
)
def orm_to_tool_test_case(orm: ExternalToolTestCase) -> ToolTestCase:
"""ORM → dataclass无敏感字段直接映射。"""
return ToolTestCase(
id=orm.id,
tool_slug=orm.tool_slug,
name=orm.name,
request_fixture=orm.request_fixture or {},
created_by=orm.created_by,
description=orm.description,
env_key=orm.env_key,
expected_status=orm.expected_status,
assertions=orm.assertions or [],
timeout=orm.timeout,
enabled=orm.enabled,
schedule_cron=orm.schedule_cron,
alert_on_failure=orm.alert_on_failure,
alert_threshold=orm.alert_threshold,
last_run_at=orm.last_run_at,
last_run_status=orm.last_run_status,
last_run_duration_ms=orm.last_run_duration_ms,
last_run_response=orm.last_run_response,
last_run_error=orm.last_run_error,
last_run_assertions_result=orm.last_run_assertions_result,
consecutive_failures=orm.consecutive_failures,
updated_by=orm.updated_by,
)
# ─── NotificationChanneldict 型敏感字段config ────────────────────────
def orm_to_notification_channel(orm: ExternalNotificationChannel) -> NotificationChannel:
"""ORM → dataclassdict 型解密 ``config``。
解密失败时抛 ``ValueError``(密钥变更场景),由调用方决定处理策略。
注意:不调用 ``orm.to_dict()``(它只返回 ``has_config`` 布尔字段)。
"""
return NotificationChannel(
id=orm.id,
channel_type=orm.channel_type,
name=orm.name,
config=decrypt_sensitive_fields(orm.config) or {},
enabled=orm.enabled,
description=orm.description,
created_by=orm.created_by,
updated_by=orm.updated_by,
created_at=orm.created_at,
updated_at=orm.updated_at,
)
def encrypt_notification_channel_for_write(data: dict[str, Any]) -> dict[str, Any]:
"""明文 dict → 密文 dictdict 型加密 ``config``。
注意:``encrypt_sensitive_fields`` 不幂等,调用方必须保证传入的是明文(未被加密过)。
"""
data = dict(data)
if data.get("config") is not None:
data["config"] = encrypt_sensitive_fields(data["config"])
return data