dify/api/services/credentials/query.py

"""Consumer-owned read contracts for credential lists and their visibility policy."""

from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from typing import Protocol

from core.plugin.entities.plugin_daemon import CredentialType


@dataclass(frozen=True, slots=True)
class ModelCredentialRecord:
    id: str
    name: str


@dataclass(frozen=True, slots=True)
class DatasourceCredentialListItem:
    id: str
    name: str
    auth_type: str
    encrypted_credentials: Mapping[str, object]
    avatar_url: str | None
    is_default: bool


@dataclass(frozen=True, slots=True)
class ToolCredentialRecord:
    id: str
    name: str
    provider: str
    credential_type: CredentialType
    credentials: Mapping[str, object]
    is_default: bool
    visibility: str
    created_by: str
    partial_member_ids: tuple[str, ...]
    from_other_member: bool


@dataclass(frozen=True, slots=True)
class TriggerSubscriptionRecord:
    id: str
    name: str
    provider: str
    credential_type: CredentialType
    credentials: Mapping[str, object]
    endpoint_id: str
    parameters: Mapping[str, object]
    properties: Mapping[str, object]
    workflows_in_use: int


class CredentialQuery(Protocol):
    """Return loaded values; implementations own SQL and short read sessions.

    An absent actor is reserved for trusted internal callers. Account-scoped
    callers must supply their actor ID; administrator roles do not bypass it.
    """

    def list_models(
        self, *, workspace_id: str, provider: str, actor_id: str | None
    ) -> Sequence[ModelCredentialRecord]: ...

    def list_datasources(
        self, *, workspace_id: str, provider: str, plugin_id: str, actor_id: str | None
    ) -> Sequence[DatasourceCredentialListItem]: ...

    def list_tools(
        self,
        *,
        workspace_id: str,
        provider: str,
        actor_id: str | None,
        include_credential_ids: Sequence[str] = (),
    ) -> Sequence[ToolCredentialRecord]: ...

    def list_trigger_subscriptions(
        self, *, workspace_id: str, provider: str, actor_id: str | None
    ) -> Sequence[TriggerSubscriptionRecord]: ...