refactor(skills): 移动SkillRepository到repositories层

解决跨模块循环导入问题,将技能仓储类从agents/skills目录下沉到通用repositories目录,统一仓储层代码位置
This commit is contained in:
Kris 2026-06-20 21:03:29 +08:00
parent bea74470cc
commit 33b7da1a43
3 changed files with 23 additions and 3 deletions

View File

@ -15,7 +15,7 @@ from langgraph.types import Command
from sqlalchemy.ext.asyncio import AsyncSession
from yuxi.agents.mcp.service import get_enabled_mcp_tools
from yuxi.agents.skills.repository import SkillRepository
from yuxi.repositories.skill_repository import SkillRepository
from yuxi.agents.skills.service import is_valid_skill_slug, list_accessible_skills, normalize_string_list
from yuxi.agents.toolkits import get_all_tool_instances
from yuxi.storage.postgres.manager import pg_manager

View File

@ -19,7 +19,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
from yuxi import config as sys_config
from yuxi.agents.mcp.service import get_enabled_mcp_server_slugs
from yuxi.agents.skills.repository import SkillRepository
from yuxi.repositories.skill_repository import SkillRepository
from yuxi.storage.postgres.models_business import Skill, User
from yuxi.utils.logging_config import logger
from yuxi.utils.share_config import SHARE_ACCESS_LEVELS, normalize_share_config

View File

@ -1,6 +1,13 @@
"""Skill 数据仓库。
``SkillRepository`` 下沉到 ``repositories`` 避免 ``external_systems/services``
直接依赖 ``agents/skills`` 上层模块导致的跨层循环导入风险
"""
from __future__ import annotations
from sqlalchemy import select
from sqlalchemy import cast, select
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.asyncio import AsyncSession
from yuxi.storage.postgres.models_business import Skill
@ -30,6 +37,19 @@ class SkillRepository:
item_map = {item.slug: item for item in items}
return [item_map[slug] for slug in normalized if slug in item_map]
async def list_by_tool_slugs(self, tool_slugs: list[str]) -> list[Skill]:
"""根据工具 slug 列表查找依赖了其中任意工具的 Skill。
使用 PostgreSQL JSONB ?| 操作符在数据库层过滤避免全表加载
"""
if not tool_slugs:
return []
stmt = select(Skill).where(
cast(Skill.tool_dependencies, JSONB).has_any(tool_slugs),
)
result = await self.db.execute(stmt)
return list(result.scalars().all())
async def get_by_slug(self, slug: str, *, for_update: bool = False) -> Skill | None:
stmt = select(Skill).where(Skill.slug == slug)
if for_update: