mirror of
https://github.com/langgenius/dify.git
synced 2026-11-11 22:02:23 +08:00
55 lines
2.2 KiB
Python
55 lines
2.2 KiB
Python
"""Owned custom API selection and native definition projection; no credentials are decrypted.
|
|
|
|
A definition is execution evidence, never a permission or a semantic read verdict.
|
|
Only the selected tenant-owned operation is projected. Native admission compares
|
|
that same operation and encrypted credential version before credential access.
|
|
"""
|
|
|
|
from collections.abc import Mapping
|
|
from typing import Any
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from core.tools.custom_tool.definition_binding import admit_api_definition, api_definition, provider_definition
|
|
from models.tools import ApiToolProvider
|
|
|
|
__all__ = ["admit_api_definition", "api_definition", "is_api_tool", "owned_api_definitions", "provider_definition"]
|
|
|
|
|
|
def owned_api_definitions(session: Session, tenant_id: str, graph: Mapping[str, Any]) -> dict[str, dict[str, Any]]:
|
|
"""Read selected tenant-owned definitions only; invalid/missing metadata grants no authority."""
|
|
selections = {
|
|
str(node["id"]): node["data"]
|
|
for node in graph.get("nodes", [])
|
|
if isinstance(node, Mapping)
|
|
and isinstance(node.get("id"), str)
|
|
and isinstance(node.get("data"), Mapping)
|
|
and is_api_tool(node["data"])
|
|
}
|
|
if not selections:
|
|
return {}
|
|
providers = session.scalars(
|
|
select(ApiToolProvider).where(
|
|
ApiToolProvider.tenant_id == tenant_id,
|
|
ApiToolProvider.id.in_({str(data.get("provider_id", "")) for data in selections.values()}),
|
|
)
|
|
).all()
|
|
rows = {row.id: row for row in providers}
|
|
result = {}
|
|
for node_id, data in selections.items():
|
|
provider = rows.get(data.get("provider_id"))
|
|
if provider is not None:
|
|
try:
|
|
definition = provider_definition(provider, str(data.get("tool_name", "")))
|
|
except (ValueError, TypeError):
|
|
definition = None
|
|
if definition is not None:
|
|
result[node_id] = definition
|
|
return result
|
|
|
|
|
|
def is_api_tool(data: Mapping[str, Any]) -> bool:
|
|
"""Recognize the native custom API capability, independent of its display name."""
|
|
return data.get("type") == "tool" and data.get("provider_type") == "api"
|