Files
dify/api/services/knowledge_fs/data_facade.py

1519 lines
55 KiB
Python

"""Typed KnowledgeFS BFF facade over operation capabilities and bounded transports."""
from __future__ import annotations
from collections.abc import Callable, Generator
from contextlib import contextmanager
from typing import Literal
from pydantic import BaseModel, JsonValue
from services.knowledge_fs.capability_broker import KnowledgeFSCapabilityBroker
from services.knowledge_fs.credential_service import KnowledgeFSServiceCredentialProfile
from services.knowledge_fs.operation_admission import KnowledgeFSOperationAdmissionService
from services.knowledge_fs.product_dto import (
KnowledgeFSAnswerTraceResponse,
KnowledgeFSBackgroundTaskListResponse,
KnowledgeFSBackgroundTaskResponse,
KnowledgeFSBulkDeletionAcceptedResponse,
KnowledgeFSBulkDocumentDeletePayload,
KnowledgeFSBulkJobResponse,
KnowledgeFSCrawlPreviewPageListResponse,
KnowledgeFSCrawlPreviewSelectionPayload,
KnowledgeFSDocumentChunkListResponse,
KnowledgeFSDocumentChunkResponse,
KnowledgeFSDocumentCompilationJobResponse,
KnowledgeFSDocumentDeletePayload,
KnowledgeFSDocumentListResponse,
KnowledgeFSDocumentMetadataPayload,
KnowledgeFSDocumentOutlineResponse,
KnowledgeFSDocumentReindexPayload,
KnowledgeFSDocumentReindexResponse,
KnowledgeFSDocumentResponse,
KnowledgeFSDocumentRevisionListResponse,
KnowledgeFSDocumentUploadAcceptedResponse,
KnowledgeFSDurableDeletionAcceptedResponse,
KnowledgeFSLogicalDocumentListResponse,
KnowledgeFSLogicalDocumentResponse,
KnowledgeFSOverviewBaseStatsResponse,
KnowledgeFSOverviewHealthResponse,
KnowledgeFSOverviewInventoryResponse,
KnowledgeFSOverviewQueryOutcomesResponse,
KnowledgeFSProfileMigrationResponse,
KnowledgeFSQueryCreatePayload,
KnowledgeFSQueryResponse,
KnowledgeFSResearchTaskCreatePayload,
KnowledgeFSResearchTaskListResponse,
KnowledgeFSResearchTaskPartialListResponse,
KnowledgeFSResearchTaskPlanPayload,
KnowledgeFSResearchTaskPlanResponse,
KnowledgeFSResearchTaskResponse,
KnowledgeFSRetrievalProfileUpdatePayload,
KnowledgeFSSettingsPayload,
KnowledgeFSSettingsResponse,
KnowledgeFSSettingsUpdateResponse,
KnowledgeFSSmallFileUploadResponse,
KnowledgeFSSourceConnectionCreatePayload,
KnowledgeFSSourceConnectionListResponse,
KnowledgeFSSourceConnectionRefreshPayload,
KnowledgeFSSourceConnectionResponse,
KnowledgeFSSourceCrawlResponse,
KnowledgeFSSourceCreatePayload,
KnowledgeFSSourceCredentialTestResponse,
KnowledgeFSSourceDeletePayload,
KnowledgeFSSourceFilesResponse,
KnowledgeFSSourceImportFilesPayload,
KnowledgeFSSourceImportPagesPayload,
KnowledgeFSSourceImportResponse,
KnowledgeFSSourceListResponse,
KnowledgeFSSourcePagesResponse,
KnowledgeFSSourceProviderListResponse,
KnowledgeFSSourceResponse,
KnowledgeFSSourceSyncPolicyPayload,
KnowledgeFSSourceSyncPolicyResponse,
KnowledgeFSSourceUpdatePayload,
KnowledgeFSSourceWorkflowCancelPayload,
KnowledgeFSSourceWorkflowImportPayload,
KnowledgeFSSourceWorkflowResponse,
KnowledgeFSSpaceUpdatePayload,
KnowledgeFSTraceEntryListResponse,
KnowledgeFSTraceListResponse,
)
from services.knowledge_fs.product_operations import KNOWLEDGE_FS_PRODUCT_OPERATIONS, is_product_operation_ready
from services.knowledge_fs.product_remote import (
KnowledgeFSOperationUnavailableError,
KnowledgeFSProductRemotePort,
KnowledgeFSProductRequestRejectedError,
KnowledgeFSRemoteBinaryRequest,
KnowledgeFSRemoteJSONRequest,
KnowledgeFSRemoteMultipartFile,
KnowledgeFSRemoteMultipartRequest,
)
class KnowledgeFSDataFacade:
def __init__(
self,
*,
admission: KnowledgeFSOperationAdmissionService,
broker: KnowledgeFSCapabilityBroker,
remote: KnowledgeFSProductRemotePort,
) -> None:
self._admission = admission
self._broker = broker
self._remote = remote
def get_settings(self, *, tenant_id: str, account_id: str, control_space_id: str) -> KnowledgeFSSettingsResponse:
raw = self._interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="getSettings",
)
return KnowledgeFSSettingsResponse.model_validate(raw)
def get_overview_stats(
self, *, tenant_id: str, account_id: str, control_space_id: str
) -> KnowledgeFSOverviewBaseStatsResponse:
raw = self._interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="getOverviewStats",
)
return KnowledgeFSOverviewBaseStatsResponse.model_validate(raw)
def get_overview_query_outcomes(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
window: Literal["24h", "7d", "30d"],
) -> KnowledgeFSOverviewQueryOutcomesResponse:
raw = self._interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="getOverviewQueryOutcomes",
query=(("window", window),),
)
return KnowledgeFSOverviewQueryOutcomesResponse.model_validate(raw)
def get_overview_inventory(
self, *, tenant_id: str, account_id: str, control_space_id: str
) -> KnowledgeFSOverviewInventoryResponse:
raw = self._interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="getOverviewInventory",
)
return KnowledgeFSOverviewInventoryResponse.model_validate(raw)
def get_overview_health(
self, *, tenant_id: str, account_id: str, control_space_id: str
) -> KnowledgeFSOverviewHealthResponse:
raw = self._interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="getOverviewHealth",
)
return KnowledgeFSOverviewHealthResponse.model_validate(raw)
def update_space(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
payload: KnowledgeFSSpaceUpdatePayload,
) -> JsonValue:
return self._interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="updateSpace",
payload=payload,
)
def update_settings(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
payload: KnowledgeFSSettingsPayload,
) -> KnowledgeFSSettingsUpdateResponse:
current = self.get_settings(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
)
if current.configuration_state == "active":
if current.revision != payload.expected_revision:
raise KnowledgeFSProductRequestRejectedError(status_code=409)
if payload.embedding is not None and payload.retrieval is not None:
raise KnowledgeFSProductRequestRejectedError(status_code=422)
if payload.embedding is not None:
migration_raw = self._interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="updateEmbeddingProfile",
payload=payload.embedding,
)
elif payload.retrieval is not None:
retrieval_revision = current.retrieval.revision if current.retrieval is not None else None
if retrieval_revision is None:
raise KnowledgeFSOperationUnavailableError(
"KnowledgeFS active retrieval profile revision is unavailable"
)
migration_raw = self._interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="updateRetrievalProfile",
payload=KnowledgeFSRetrievalProfileUpdatePayload(
expectedRevision=retrieval_revision,
profile=payload.retrieval,
),
)
else:
raise KnowledgeFSProductRequestRejectedError(status_code=422)
return KnowledgeFSSettingsUpdateResponse(
migration=KnowledgeFSProfileMigrationResponse.model_validate(migration_raw),
settings=current,
)
raw = self._interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="updateSettings",
payload=payload,
)
return KnowledgeFSSettingsUpdateResponse(settings=KnowledgeFSSettingsResponse.model_validate(raw))
def get_profile_migration(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
migration_id: str,
) -> KnowledgeFSProfileMigrationResponse:
raw = self._interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="getProfileMigration",
path_parameters=(("migrationId", migration_id),),
)
return KnowledgeFSProfileMigrationResponse.model_validate(raw)
def list_documents(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
cursor: str | None,
) -> KnowledgeFSDocumentListResponse:
raw = self._interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="listDocuments",
query=(("cursor", cursor),) if cursor else (),
)
return KnowledgeFSDocumentListResponse.model_validate(raw)
def list_logical_documents(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
cursor: str | None,
) -> KnowledgeFSLogicalDocumentListResponse:
raw = self._interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="listLogicalDocuments",
query=(("cursor", cursor),) if cursor else (),
)
return KnowledgeFSLogicalDocumentListResponse.model_validate(raw)
def get_logical_document(
self, *, tenant_id: str, account_id: str, control_space_id: str, document_id: str
) -> KnowledgeFSLogicalDocumentResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="getLogicalDocument",
resource_id=document_id,
path_parameters=(("documentId", document_id),),
)
return KnowledgeFSLogicalDocumentResponse.model_validate(raw)
def create_document(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
body_reader: Callable[[int], KnowledgeFSRemoteMultipartFile],
) -> KnowledgeFSDocumentUploadAcceptedResponse:
operation_id = "createDocument"
_assert_multipart_bff_ready(operation_id)
with self._admitted(tenant_id=tenant_id, operation_id=operation_id):
issued = self._broker.issue_interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id=operation_id,
)
operation = KNOWLEDGE_FS_PRODUCT_OPERATIONS[operation_id]
upload = body_reader(operation.max_request_bytes)
if not upload.filename or not upload.content_type or not isinstance(upload.body, bytes) or not upload.body:
raise KnowledgeFSProductRequestRejectedError(status_code=422)
if len(upload.body) > operation.max_request_bytes:
raise KnowledgeFSProductRequestRejectedError(status_code=413)
if operation.kfs_path is None:
raise KnowledgeFSOperationUnavailableError(f"KnowledgeFS operation is unavailable: {operation_id}")
path = _resolve_product_path(
template=operation.kfs_path,
knowledge_space_id=issued.knowledge_space_id,
resource_id=None,
resource_resolver=operation.resource_resolver,
path_parameters=(),
)
raw = self._remote.execute_multipart(
KnowledgeFSRemoteMultipartRequest(
operation_id=operation_id,
method=operation.method,
path=path,
namespace_id=tenant_id,
knowledge_space_id=issued.knowledge_space_id,
capability_token=issued.token,
trace_id=issued.trace_id,
file=upload,
)
)
return KnowledgeFSDocumentUploadAcceptedResponse.model_validate(raw)
def upload_small_file(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
upload_session_id: str,
body_reader: Callable[[int], bytes],
) -> KnowledgeFSSmallFileUploadResponse:
operation_id = "uploadSmallFile"
_assert_binary_bff_ready(operation_id)
with self._admitted(tenant_id=tenant_id, operation_id=operation_id):
issued = self._broker.issue_interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id=operation_id,
resource_id=upload_session_id,
)
operation = KNOWLEDGE_FS_PRODUCT_OPERATIONS[operation_id]
body = body_reader(operation.max_request_bytes)
if not isinstance(body, bytes) or not body:
raise KnowledgeFSProductRequestRejectedError(status_code=422)
if len(body) > operation.max_request_bytes:
raise KnowledgeFSProductRequestRejectedError(status_code=413)
if operation.kfs_path is None:
raise KnowledgeFSOperationUnavailableError(f"KnowledgeFS operation is unavailable: {operation_id}")
path = _resolve_product_path(
template=operation.kfs_path,
knowledge_space_id=issued.knowledge_space_id,
resource_id=upload_session_id,
resource_resolver=operation.resource_resolver,
path_parameters=(),
)
raw = self._remote.execute_binary(
KnowledgeFSRemoteBinaryRequest(
operation_id=operation_id,
method=operation.method,
path=path,
namespace_id=tenant_id,
knowledge_space_id=issued.knowledge_space_id,
capability_token=issued.token,
trace_id=issued.trace_id,
body=body,
query=(("knowledgeSpaceId", issued.knowledge_space_id),),
)
)
return KnowledgeFSSmallFileUploadResponse.model_validate(raw)
def get_document(
self, *, tenant_id: str, account_id: str, control_space_id: str, document_id: str
) -> KnowledgeFSDocumentResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="getDocument",
resource_id=document_id,
path_parameters=(("documentId", document_id),),
)
return KnowledgeFSDocumentResponse.model_validate(raw)
def get_document_outline(
self, *, tenant_id: str, account_id: str, control_space_id: str, document_id: str
) -> KnowledgeFSDocumentOutlineResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="getDocumentOutline",
resource_id=document_id,
path_parameters=(("documentId", document_id),),
)
return KnowledgeFSDocumentOutlineResponse.model_validate(raw)
def list_document_revisions(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
document_id: str,
cursor: str | None = None,
) -> KnowledgeFSDocumentRevisionListResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="listDocumentRevisions",
resource_id=document_id,
path_parameters=(("documentId", document_id),),
query=(("cursor", cursor),) if cursor else (),
)
return KnowledgeFSDocumentRevisionListResponse.model_validate(raw)
def update_document_metadata(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
document_id: str,
payload: KnowledgeFSDocumentMetadataPayload,
) -> KnowledgeFSLogicalDocumentResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="updateDocumentMetadata",
resource_id=document_id,
path_parameters=(("documentId", document_id),),
payload=payload,
)
return KnowledgeFSLogicalDocumentResponse.model_validate(raw)
def list_document_chunks(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
document_id: str,
revision: int,
cursor: str | None = None,
query_text: str | None = None,
) -> KnowledgeFSDocumentChunkListResponse:
query = tuple((name, value) for name, value in (("cursor", cursor), ("query", query_text)) if value)
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="listDocumentChunks",
resource_id=document_id,
path_parameters=(("documentId", document_id), ("revision", str(revision))),
query=query,
)
return KnowledgeFSDocumentChunkListResponse.model_validate(raw)
def get_document_chunk(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
document_id: str,
revision: int,
chunk_id: str,
) -> KnowledgeFSDocumentChunkResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="getDocumentChunk",
resource_id=document_id,
path_parameters=(
("documentId", document_id),
("revision", str(revision)),
("chunkId", chunk_id),
),
)
return KnowledgeFSDocumentChunkResponse.model_validate(raw)
def delete_document(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
document_id: str,
payload: KnowledgeFSDocumentDeletePayload,
idempotency_key: str,
) -> KnowledgeFSDurableDeletionAcceptedResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="deleteDocument",
resource_id=document_id,
path_parameters=(("documentId", document_id),),
payload=payload,
headers=(("Idempotency-Key", idempotency_key),),
)
return KnowledgeFSDurableDeletionAcceptedResponse.model_validate(raw)
def bulk_delete_documents(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
payload: KnowledgeFSBulkDocumentDeletePayload,
idempotency_key: str,
) -> KnowledgeFSBulkDeletionAcceptedResponse:
raw = self._interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="bulkDeleteDocuments",
payload=payload,
headers=(("Idempotency-Key", idempotency_key),),
)
return KnowledgeFSBulkDeletionAcceptedResponse.model_validate(raw)
def reindex_documents(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
payload: KnowledgeFSDocumentReindexPayload,
) -> KnowledgeFSDocumentReindexResponse:
raw = self._interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="reindexDocuments",
payload=payload,
)
return KnowledgeFSDocumentReindexResponse.model_validate(raw)
def get_compilation_job(
self, *, tenant_id: str, account_id: str, control_space_id: str, job_id: str
) -> KnowledgeFSDocumentCompilationJobResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="getCompilationJob",
resource_id=job_id,
)
return KnowledgeFSDocumentCompilationJobResponse.model_validate(raw)
def cancel_compilation_job(
self, *, tenant_id: str, account_id: str, control_space_id: str, job_id: str
) -> KnowledgeFSDocumentCompilationJobResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="cancelCompilationJob",
resource_id=job_id,
)
return KnowledgeFSDocumentCompilationJobResponse.model_validate(raw)
def retry_compilation_job(
self, *, tenant_id: str, account_id: str, control_space_id: str, job_id: str
) -> KnowledgeFSDocumentCompilationJobResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="retryCompilationJob",
resource_id=job_id,
)
return KnowledgeFSDocumentCompilationJobResponse.model_validate(raw)
def get_bulk_job(
self, *, tenant_id: str, account_id: str, control_space_id: str, job_id: str
) -> KnowledgeFSBulkJobResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="getBulkJob",
resource_id=job_id,
)
return KnowledgeFSBulkJobResponse.model_validate(raw)
def list_background_tasks(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
cursor: str | None = None,
limit: int = 50,
) -> KnowledgeFSBackgroundTaskListResponse:
query = (("limit", str(limit)),) + ((("cursor", cursor),) if cursor else ())
raw = self._interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="listBackgroundTasks",
query=query,
)
return KnowledgeFSBackgroundTaskListResponse.model_validate(raw)
def cancel_background_task(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
task_kind: Literal["document", "document_bulk", "source"],
task_id: str,
) -> KnowledgeFSBackgroundTaskResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="cancelBackgroundTask",
resource_id=task_id,
path_parameters=(("taskKind", task_kind), ("taskId", task_id)),
)
return KnowledgeFSBackgroundTaskResponse.model_validate(raw)
def retry_background_task(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
task_kind: Literal["document", "document_bulk", "source"],
task_id: str,
) -> KnowledgeFSBackgroundTaskResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="retryBackgroundTask",
resource_id=task_id,
path_parameters=(("taskKind", task_kind), ("taskId", task_id)),
)
return KnowledgeFSBackgroundTaskResponse.model_validate(raw)
def list_sources(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
cursor: str | None = None,
limit: int = 50,
) -> KnowledgeFSSourceListResponse:
raw = self._interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="listSources",
query=(
*((("cursor", cursor),) if cursor else ()),
("limit", str(limit)),
),
)
return KnowledgeFSSourceListResponse.model_validate(raw)
def create_source(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
payload: KnowledgeFSSourceCreatePayload,
) -> KnowledgeFSSourceResponse:
raw = self._interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="createSource",
payload=payload,
)
return KnowledgeFSSourceResponse.model_validate(raw)
def get_source(
self, *, tenant_id: str, account_id: str, control_space_id: str, source_id: str
) -> KnowledgeFSSourceResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="getSource",
resource_id=source_id,
path_parameters=(("sourceId", source_id),),
)
return KnowledgeFSSourceResponse.model_validate(raw)
def update_source(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
source_id: str,
payload: KnowledgeFSSourceUpdatePayload,
) -> KnowledgeFSSourceResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="updateSource",
resource_id=source_id,
path_parameters=(("sourceId", source_id),),
payload=payload,
)
return KnowledgeFSSourceResponse.model_validate(raw)
def delete_source(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
source_id: str,
payload: KnowledgeFSSourceDeletePayload,
documents: Literal["cascade", "keep"],
idempotency_key: str,
) -> KnowledgeFSDurableDeletionAcceptedResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="deleteSource",
resource_id=source_id,
path_parameters=(("sourceId", source_id),),
payload=payload,
query=(("documents", documents),),
headers=(("Idempotency-Key", idempotency_key),),
)
return KnowledgeFSDurableDeletionAcceptedResponse.model_validate(raw)
def test_source(
self, *, tenant_id: str, account_id: str, control_space_id: str, source_id: str
) -> KnowledgeFSSourceCredentialTestResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="testSource",
resource_id=source_id,
path_parameters=(("sourceId", source_id),),
)
return KnowledgeFSSourceCredentialTestResponse.model_validate(raw)
def sync_source(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
source_id: str,
idempotency_key: str,
) -> KnowledgeFSSourceWorkflowResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="syncSource",
resource_id=source_id,
path_parameters=(("sourceId", source_id),),
headers=(("Idempotency-Key", idempotency_key),),
)
return KnowledgeFSSourceWorkflowResponse.model_validate(raw)
def list_source_providers(
self, *, tenant_id: str, account_id: str, control_space_id: str
) -> KnowledgeFSSourceProviderListResponse:
raw = self._interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="listSourceProviders",
)
return KnowledgeFSSourceProviderListResponse.model_validate(raw)
def create_source_connection(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
payload: KnowledgeFSSourceConnectionCreatePayload,
) -> KnowledgeFSSourceConnectionResponse:
raw = self._interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="createSourceConnection",
payload=payload,
)
return KnowledgeFSSourceConnectionResponse.model_validate(raw)
def list_source_connections(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
cursor: str | None,
limit: int,
) -> KnowledgeFSSourceConnectionListResponse:
query = (("limit", str(limit)),) + ((("cursor", cursor),) if cursor else ())
raw = self._interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="listSourceConnections",
query=query,
)
return KnowledgeFSSourceConnectionListResponse.model_validate(raw)
def refresh_source_connection(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
connection_id: str,
payload: KnowledgeFSSourceConnectionRefreshPayload,
) -> KnowledgeFSSourceConnectionResponse:
raw = self._interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="refreshSourceConnection",
payload=payload,
path_parameters=(("connectionId", connection_id),),
)
return KnowledgeFSSourceConnectionResponse.model_validate(raw)
def preview_source_crawl(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
source_id: str,
idempotency_key: str,
) -> KnowledgeFSSourceWorkflowResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="previewSourceCrawl",
resource_id=source_id,
path_parameters=(("sourceId", source_id),),
headers=(("Idempotency-Key", idempotency_key),),
)
return KnowledgeFSSourceWorkflowResponse.model_validate(raw)
def import_source_workflow(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
source_id: str,
payload: KnowledgeFSSourceWorkflowImportPayload,
idempotency_key: str,
) -> KnowledgeFSSourceWorkflowResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="importSourceWorkflow",
resource_id=source_id,
path_parameters=(("sourceId", source_id),),
payload=payload,
headers=(("Idempotency-Key", idempotency_key),),
)
return KnowledgeFSSourceWorkflowResponse.model_validate(raw)
def get_source_sync_policy(
self, *, tenant_id: str, account_id: str, control_space_id: str, source_id: str
) -> KnowledgeFSSourceSyncPolicyResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="getSourceSyncPolicy",
resource_id=source_id,
path_parameters=(("sourceId", source_id),),
)
return KnowledgeFSSourceSyncPolicyResponse.model_validate(raw)
def update_source_sync_policy(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
source_id: str,
payload: KnowledgeFSSourceSyncPolicyPayload,
) -> KnowledgeFSSourceSyncPolicyResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="updateSourceSyncPolicy",
resource_id=source_id,
path_parameters=(("sourceId", source_id),),
payload=payload,
)
return KnowledgeFSSourceSyncPolicyResponse.model_validate(raw)
def get_source_workflow(
self, *, tenant_id: str, account_id: str, control_space_id: str, run_id: str
) -> KnowledgeFSSourceWorkflowResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="getSourceWorkflow",
resource_id=run_id,
path_parameters=(("runId", run_id),),
)
return KnowledgeFSSourceWorkflowResponse.model_validate(raw)
def cancel_source_workflow(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
run_id: str,
payload: KnowledgeFSSourceWorkflowCancelPayload,
) -> KnowledgeFSSourceWorkflowResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="cancelSourceWorkflow",
resource_id=run_id,
path_parameters=(("runId", run_id),),
payload=payload,
)
return KnowledgeFSSourceWorkflowResponse.model_validate(raw)
def retry_source_workflow(
self, *, tenant_id: str, account_id: str, control_space_id: str, run_id: str
) -> KnowledgeFSSourceWorkflowResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="retrySourceWorkflow",
resource_id=run_id,
path_parameters=(("runId", run_id),),
)
return KnowledgeFSSourceWorkflowResponse.model_validate(raw)
def list_crawl_preview_pages(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
run_id: str,
cursor: str | None,
limit: int,
) -> KnowledgeFSCrawlPreviewPageListResponse:
query = (("limit", str(limit)),) + ((("cursor", cursor),) if cursor else ())
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="listCrawlPreviewPages",
resource_id=run_id,
path_parameters=(("runId", run_id),),
query=query,
)
return KnowledgeFSCrawlPreviewPageListResponse.model_validate(raw)
def select_crawl_preview_pages(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
run_id: str,
payload: KnowledgeFSCrawlPreviewSelectionPayload,
idempotency_key: str,
) -> KnowledgeFSSourceWorkflowResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="selectCrawlPreviewPages",
resource_id=run_id,
path_parameters=(("runId", run_id),),
payload=payload,
headers=(("Idempotency-Key", idempotency_key),),
)
return KnowledgeFSSourceWorkflowResponse.model_validate(raw)
def crawl_source(
self, *, tenant_id: str, account_id: str, control_space_id: str, source_id: str
) -> KnowledgeFSSourceCrawlResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="crawlSource",
resource_id=source_id,
path_parameters=(("sourceId", source_id),),
)
return KnowledgeFSSourceCrawlResponse.model_validate(raw)
def list_source_pages(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
source_id: str,
cursor: str | None = None,
limit: int = 50,
) -> KnowledgeFSSourcePagesResponse:
query = (("limit", str(limit)),) + ((("cursor", cursor),) if cursor else ())
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="listSourcePages",
resource_id=source_id,
path_parameters=(("sourceId", source_id),),
query=query,
)
return KnowledgeFSSourcePagesResponse.model_validate(raw)
def import_source_pages(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
source_id: str,
payload: KnowledgeFSSourceImportPagesPayload,
) -> KnowledgeFSSourceImportResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="importSourcePages",
resource_id=source_id,
path_parameters=(("sourceId", source_id),),
payload=payload,
)
return KnowledgeFSSourceImportResponse.model_validate(raw)
def list_source_files(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
source_id: str,
query: tuple[tuple[str, str], ...] = (),
) -> KnowledgeFSSourceFilesResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="listSourceFiles",
resource_id=source_id,
path_parameters=(("sourceId", source_id),),
query=query,
)
return KnowledgeFSSourceFilesResponse.model_validate(raw)
def import_source_files(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
source_id: str,
payload: KnowledgeFSSourceImportFilesPayload,
) -> KnowledgeFSSourceImportResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="importSourceFiles",
resource_id=source_id,
path_parameters=(("sourceId", source_id),),
payload=payload,
)
return KnowledgeFSSourceImportResponse.model_validate(raw)
def create_query(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
payload: KnowledgeFSQueryCreatePayload,
) -> KnowledgeFSQueryResponse:
_ = (tenant_id, account_id, control_space_id, payload)
raise KnowledgeFSOperationUnavailableError(
"Buffered KnowledgeFS query creation is deprecated; use the queries/admission direct flow"
)
def create_research_task(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
payload: KnowledgeFSResearchTaskCreatePayload,
) -> KnowledgeFSResearchTaskResponse:
raw = self._interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="createResearchTask",
payload=payload,
bind_space_in_body=True,
)
return KnowledgeFSResearchTaskResponse.model_validate(raw)
def plan_research_task(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
payload: KnowledgeFSResearchTaskPlanPayload,
) -> KnowledgeFSResearchTaskPlanResponse:
raw = self._interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="planResearchTask",
payload=payload,
bind_space_in_body=True,
)
return KnowledgeFSResearchTaskPlanResponse.model_validate(raw)
def get_research_task(
self, *, tenant_id: str, account_id: str, control_space_id: str, task_id: str
) -> KnowledgeFSResearchTaskResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="getResearchTask",
resource_id=task_id,
)
return KnowledgeFSResearchTaskResponse.model_validate(raw)
def list_research_task_partials(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
task_id: str,
cursor: str | None = None,
limit: int = 25,
) -> KnowledgeFSResearchTaskPartialListResponse:
query = (("limit", str(limit)),) + ((("cursor", cursor),) if cursor else ())
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="listResearchTaskPartials",
resource_id=task_id,
query=query,
)
return KnowledgeFSResearchTaskPartialListResponse.model_validate(raw)
def cancel_research_task(
self, *, tenant_id: str, account_id: str, control_space_id: str, task_id: str
) -> KnowledgeFSResearchTaskResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="cancelResearchTask",
resource_id=task_id,
)
return KnowledgeFSResearchTaskResponse.model_validate(raw)
def list_research_tasks(
self, *, tenant_id: str, account_id: str, control_space_id: str, cursor: str | None = None
) -> KnowledgeFSResearchTaskListResponse:
raw = self._interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="listResearchTasks",
query=(("cursor", cursor),) if cursor else (),
)
return KnowledgeFSResearchTaskListResponse.model_validate(raw)
def list_traces(
self, *, tenant_id: str, account_id: str, control_space_id: str, cursor: str | None = None
) -> KnowledgeFSTraceListResponse:
raw = self._interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="listTraces",
query=(("cursor", cursor),) if cursor else (),
)
return KnowledgeFSTraceListResponse.model_validate(raw)
def get_trace(
self, *, tenant_id: str, account_id: str, control_space_id: str, trace_id: str
) -> KnowledgeFSAnswerTraceResponse:
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id="getTrace",
resource_id=trace_id,
path_parameters=(("traceId", trace_id),),
)
return KnowledgeFSAnswerTraceResponse.model_validate(raw)
def list_trace_entries(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
trace_id: str,
kind: Literal["conflicts", "evidence", "missing"],
cursor: str | None = None,
limit: int = 100,
) -> KnowledgeFSTraceEntryListResponse:
operation_id = {
"conflicts": "listTraceConflicts",
"evidence": "listTraceEvidence",
"missing": "listTraceMissing",
}[kind]
query = (("limit", str(limit)),) + ((("cursor", cursor),) if cursor else ())
raw = self._interactive_child(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id=operation_id,
resource_id=trace_id,
path_parameters=(("traceId", trace_id),),
query=query,
)
return KnowledgeFSTraceEntryListResponse.model_validate(raw)
def execute_service(
self,
*,
profile: KnowledgeFSServiceCredentialProfile,
operation_id: str,
payload: BaseModel | None = None,
query: tuple[tuple[str, str], ...] = (),
bind_space_in_body: bool = False,
resource_id: str | None = None,
path_parameters: tuple[tuple[str, str], ...] = (),
headers: tuple[tuple[str, str], ...] = (),
) -> JsonValue:
_assert_json_bff_ready(operation_id)
with self._admitted(tenant_id=profile.tenant_id, operation_id=operation_id):
issued = self._broker.issue_service(profile=profile, operation_id=operation_id, resource_id=resource_id)
return self._execute(
operation_id=operation_id,
namespace_id=profile.tenant_id,
knowledge_space_id=profile.knowledge_space_id,
knowledge_space_revision=profile.knowledge_space_revision,
capability_token=issued.token,
trace_id=issued.trace_id,
payload=payload,
query=query,
bind_space_in_body=bind_space_in_body,
resource_id=resource_id,
path_parameters=path_parameters,
headers=headers,
)
def _interactive(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
operation_id: str,
payload: BaseModel | None = None,
query: tuple[tuple[str, str], ...] = (),
bind_space_in_body: bool = False,
resource_id: str | None = None,
path_parameters: tuple[tuple[str, str], ...] = (),
headers: tuple[tuple[str, str], ...] = (),
) -> JsonValue:
_assert_json_bff_ready(operation_id)
with self._admitted(tenant_id=tenant_id, operation_id=operation_id):
issued = self._broker.issue_interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id=operation_id,
resource_id=resource_id,
)
return self._execute(
operation_id=operation_id,
namespace_id=tenant_id,
knowledge_space_id=issued.knowledge_space_id,
knowledge_space_revision=issued.knowledge_space_revision,
capability_token=issued.token,
trace_id=issued.trace_id,
payload=payload,
query=query,
bind_space_in_body=bind_space_in_body,
resource_id=resource_id,
path_parameters=path_parameters,
headers=headers,
)
@contextmanager
def _admitted(self, *, tenant_id: str, operation_id: str) -> Generator[None]:
charge = self._admission.reserve(tenant_id=tenant_id, operation_id=operation_id)
try:
yield
except BaseException:
charge.refund()
raise
else:
charge.commit()
def _interactive_child(
self,
*,
tenant_id: str,
account_id: str,
control_space_id: str,
operation_id: str,
resource_id: str,
payload: BaseModel | None = None,
query: tuple[tuple[str, str], ...] = (),
path_parameters: tuple[tuple[str, str], ...] = (),
headers: tuple[tuple[str, str], ...] = (),
) -> JsonValue:
return self._interactive(
tenant_id=tenant_id,
account_id=account_id,
control_space_id=control_space_id,
operation_id=operation_id,
resource_id=resource_id,
payload=payload,
query=query,
path_parameters=path_parameters,
headers=headers,
)
def _execute(
self,
*,
operation_id: str,
namespace_id: str,
knowledge_space_id: str,
knowledge_space_revision: int,
capability_token: str,
trace_id: str,
payload: BaseModel | None,
query: tuple[tuple[str, str], ...],
bind_space_in_body: bool,
resource_id: str | None,
path_parameters: tuple[tuple[str, str], ...],
headers: tuple[tuple[str, str], ...],
) -> JsonValue:
operation = KNOWLEDGE_FS_PRODUCT_OPERATIONS[operation_id]
if operation.kfs_path is None:
raise KnowledgeFSOperationUnavailableError(f"KnowledgeFS operation is unavailable: {operation_id}")
path = _resolve_product_path(
template=operation.kfs_path,
knowledge_space_id=knowledge_space_id,
resource_id=resource_id,
resource_resolver=operation.resource_resolver,
path_parameters=path_parameters,
)
remote_payload: JsonValue | None = (
payload.model_dump(mode="json", exclude_none=True, by_alias=True) if payload is not None else None
)
if operation_id == "updateSpace":
if knowledge_space_revision <= 0 or not isinstance(payload, KnowledgeFSSpaceUpdatePayload):
raise KnowledgeFSOperationUnavailableError("KnowledgeFS metadata revision is not available")
remote_payload = {"expectedRevision": knowledge_space_revision}
if payload.name is not None:
remote_payload["name"] = payload.name
if payload.icon is not None:
remote_payload["iconRef"] = payload.icon
if payload.description is not None:
remote_payload["description"] = payload.description
if bind_space_in_body:
if not isinstance(remote_payload, dict):
raise KnowledgeFSOperationUnavailableError("KnowledgeFS request body binding is unavailable")
remote_payload["knowledgeSpaceId"] = knowledge_space_id
if operation.resource_resolver in {"job", "query", "research_task"} and not path.startswith(
"/knowledge-spaces/"
):
query = (*query, ("knowledgeSpaceId", knowledge_space_id))
return self._remote.execute_json(
KnowledgeFSRemoteJSONRequest(
operation_id=operation_id,
method=operation.method,
path=path,
namespace_id=namespace_id,
knowledge_space_id=knowledge_space_id,
capability_token=capability_token,
trace_id=trace_id,
payload=remote_payload,
query=query,
headers=headers,
)
)
def _resolve_product_path(
*,
template: str,
knowledge_space_id: str,
resource_id: str | None,
resource_resolver: str,
path_parameters: tuple[tuple[str, str], ...],
) -> str:
bindings = {name: _path_segment(value) for name, value in path_parameters}
if template.startswith("/knowledge-spaces/{id}") or resource_resolver in {"knowledge_space", "namespace"}:
bindings["id"] = _path_segment(knowledge_space_id)
elif "{id}" in template:
if resource_id is None:
raise KnowledgeFSOperationUnavailableError("KnowledgeFS child resource is required")
bindings["id"] = _path_segment(resource_id)
path = template
for name, value in bindings.items():
path = path.replace(f"{{{name}}}", value)
if "{" in path or "}" in path:
raise KnowledgeFSOperationUnavailableError("KnowledgeFS product path resolver is not registered")
return path
def _path_segment(value: str) -> str:
normalized = value.strip()
if (
not normalized
or len(normalized) > 1_000
or normalized in {".", ".."}
or any(character in normalized for character in ("/", "%", "?", "#", "\\"))
):
raise KnowledgeFSOperationUnavailableError("KnowledgeFS product path parameter is invalid")
return normalized
def _assert_ready(operation_id: str) -> None:
if not is_product_operation_ready(operation_id):
raise KnowledgeFSOperationUnavailableError(f"KnowledgeFS operation is unavailable: {operation_id}")
def _assert_json_bff_ready(operation_id: str) -> None:
_assert_ready(operation_id)
if KNOWLEDGE_FS_PRODUCT_OPERATIONS[operation_id].transport != "json":
raise KnowledgeFSOperationUnavailableError(
f"KnowledgeFS operation requires a direct transport and cannot use the buffered BFF: {operation_id}"
)
def _assert_binary_bff_ready(operation_id: str) -> None:
_assert_ready(operation_id)
if KNOWLEDGE_FS_PRODUCT_OPERATIONS[operation_id].transport != "binary":
raise KnowledgeFSOperationUnavailableError(
f"KnowledgeFS operation does not allow the bounded binary BFF: {operation_id}"
)
def _assert_multipart_bff_ready(operation_id: str) -> None:
_assert_ready(operation_id)
if KNOWLEDGE_FS_PRODUCT_OPERATIONS[operation_id].transport != "multipart":
raise KnowledgeFSOperationUnavailableError(
f"KnowledgeFS operation does not allow the bounded multipart BFF: {operation_id}"
)
__all__ = ["KnowledgeFSDataFacade"]