Kris
1965b9f742
1. 将工具、webhook的时间DTO字段从字符串改为datetime类型 2. 移除工具服务中的手动ISO时间解析逻辑 3. 新增CapabilityNotSupportedError异常与get_integration_or_raise方法 4. 格式化代码行内表达式简化写法
138 lines
5.5 KiB
Python
138 lines
5.5 KiB
Python
"""IntegrationRegistry：厂商集成目录注册表。
|
||
|
||
实现 ``core/contracts.py::IntegrationRegistry`` 端口（结构化子类型，无需显式继承）。
|
||
|
||
维护 ``key -> IntegrationMetadata`` 映射，应用启动时由各集成包 ``__init__.py``
|
||
通过 ``register()`` 自注册，运行时只读。元数据为内存态，不持久化到数据库。
|
||
``clear()`` 仅供测试隔离使用。
|
||
|
||
键冲突策略：同一 ``key`` 重复注册且元数据实例不同时抛 ``DomainValidationError``，
|
||
与 framework 层 ``register_protocol_auth_plugin`` / ``http_error_extractor_registry``
|
||
等注册表策略一致，避免静默覆盖。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from yuxi.external_systems.exceptions import DomainValidationError, EntityNotFoundError
|
||
from yuxi.external_systems.integrations.schemas import IntegrationMetadata
|
||
|
||
|
||
class IntegrationRegistry:
|
||
"""厂商集成目录注册表，实现 core/contracts.py::IntegrationRegistry 端口。
|
||
|
||
实现 Protocol 的 3 个查询方法（list_integrations / get_integration /
|
||
list_adapter_types），并额外提供 register / clear 供框架内部使用
|
||
（与 framework/registry/adapter_registry.py::AdapterRegistry 模式一致）。
|
||
"""
|
||
|
||
_integrations: dict[str, IntegrationMetadata] = {}
|
||
|
||
@classmethod
|
||
def register(cls, metadata: IntegrationMetadata) -> None:
|
||
"""注册厂商集成元数据。键冲突抛 DomainValidationError。"""
|
||
existing = cls._integrations.get(metadata.key)
|
||
if existing is not None and existing is not metadata:
|
||
raise DomainValidationError(
|
||
f"厂商集成元数据冲突: key={metadata.key} "
|
||
f"已注册 {existing.display_name}，尝试注册 {metadata.display_name}"
|
||
)
|
||
cls._integrations[metadata.key] = metadata
|
||
|
||
@classmethod
|
||
def filter_integrations(
|
||
cls,
|
||
*,
|
||
adapter_type: str | None = None,
|
||
tags: list[str] | None = None,
|
||
keyword: str | None = None,
|
||
) -> list[IntegrationMetadata]:
|
||
"""按条件过滤并稳定排序厂商集成（不分页）。
|
||
|
||
list_integrations 与 count_integrations 的共用底层方法，
|
||
保证过滤逻辑单一来源，避免两处实现漂移。overview / categories
|
||
端点需要全量过滤结果时也直接调用本方法。
|
||
|
||
- adapter_type: 精确匹配 metadata.adapter_type
|
||
- tags: OR 关系（命中任一标签即返回）
|
||
- keyword: 模糊匹配 display_name / description（大小写不敏感）
|
||
- 按 key 升序稳定排序，保证分页结果可重现
|
||
"""
|
||
items = list(cls._integrations.values())
|
||
if adapter_type is not None:
|
||
items = [it for it in items if it.adapter_type == adapter_type]
|
||
if tags:
|
||
tag_set = set(tags)
|
||
items = [it for it in items if tag_set.intersection(it.tags)]
|
||
if keyword:
|
||
kw = keyword.lower()
|
||
items = [it for it in items if kw in it.display_name.lower() or kw in it.description.lower()]
|
||
items.sort(key=lambda it: it.key)
|
||
return items
|
||
|
||
@classmethod
|
||
def list_integrations(
|
||
cls,
|
||
*,
|
||
adapter_type: str | None = None,
|
||
tags: list[str] | None = None,
|
||
keyword: str | None = None,
|
||
limit: int = 50,
|
||
offset: int = 0,
|
||
) -> list[IntegrationMetadata]:
|
||
"""查询厂商集成目录，支持按适配器类型/标签/关键词过滤与分页。
|
||
|
||
分页基于 filter_integrations 的全量结果切片，全量总数请用
|
||
count_integrations 获取（list 返回的 len(items) 仅为当前页条数）。
|
||
"""
|
||
filtered = cls.filter_integrations(adapter_type=adapter_type, tags=tags, keyword=keyword)
|
||
if offset < 0:
|
||
offset = 0
|
||
if limit < 0:
|
||
limit = 0
|
||
return filtered[offset : offset + limit]
|
||
|
||
@classmethod
|
||
def count_integrations(
|
||
cls,
|
||
*,
|
||
adapter_type: str | None = None,
|
||
tags: list[str] | None = None,
|
||
keyword: str | None = None,
|
||
) -> int:
|
||
"""统计满足筛选条件的厂商集成全量总数（不分页）。
|
||
|
||
与 list_integrations 共用 filter_integrations，保证 total 与
|
||
分页结果一致。供 Router 层返回真实全量 total，避免前端分页页数错误。
|
||
"""
|
||
return len(cls.filter_integrations(adapter_type=adapter_type, tags=tags, keyword=keyword))
|
||
|
||
@classmethod
|
||
def get_integration(cls, key: str) -> IntegrationMetadata | None:
|
||
"""按 key 获取厂商集成元数据，未找到返回 None。"""
|
||
return cls._integrations.get(key)
|
||
|
||
@classmethod
|
||
def get_integration_or_raise(cls, key: str) -> IntegrationMetadata:
|
||
"""按 key 获取厂商集成元数据，未找到抛 ``EntityNotFoundError`` (404)。
|
||
|
||
供 Router 的详情类端点使用，避免每个端点重复 None 判断与抛异常逻辑。
|
||
"""
|
||
metadata = cls._integrations.get(key)
|
||
if metadata is None:
|
||
raise EntityNotFoundError(f"厂商集成 {key} 不存在")
|
||
return metadata
|
||
|
||
@classmethod
|
||
def list_adapter_types(cls) -> list[str]:
|
||
"""列出所有已注册厂商集成涉及的适配器类型（去重）。"""
|
||
seen: dict[str, None] = {}
|
||
for it in cls._integrations.values():
|
||
if it.adapter_type not in seen:
|
||
seen[it.adapter_type] = None
|
||
return list(seen.keys())
|
||
|
||
@classmethod
|
||
def clear(cls) -> None:
|
||
"""清空注册表，仅供测试使用。"""
|
||
cls._integrations.clear()
|