139 lines
4.6 KiB
Python
139 lines
4.6 KiB
Python
"""Persistence contract for DeerFlow-owned LLMWiki authorization metadata."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from abc import ABC, abstractmethod
|
|
from typing import Any, Literal
|
|
|
|
LlmWikiScope = Literal["personal", "public", "all"]
|
|
UNSET = object()
|
|
|
|
|
|
def serialize_wiki_parent_directories(
|
|
value: list[str] | None,
|
|
) -> list[str] | None:
|
|
return None if value is None else [str(item) for item in value]
|
|
|
|
|
|
def serialize_wiki_directory_paths(
|
|
value: list[list[str]] | None,
|
|
) -> list[list[str]] | None:
|
|
return (
|
|
None
|
|
if value is None
|
|
else [[str(segment) for segment in path] for path in value]
|
|
)
|
|
|
|
|
|
class LlmWikiStore(ABC):
|
|
@abstractmethod
|
|
async def create_mapping(
|
|
self,
|
|
*,
|
|
weknora_id: str,
|
|
owner_user_id: str,
|
|
name: str,
|
|
description: str,
|
|
kb_type: str,
|
|
retrieval_settings: dict[str, Any] | None = None,
|
|
wiki_parent_directories: list[str] | None = None,
|
|
wiki_directory_paths: list[list[str]] | None = None,
|
|
wiki_focus_instructions: str | None = None,
|
|
wiki_content_instructions_raw: str | None = None,
|
|
wiki_extraction_instructions_raw: str | None = None,
|
|
wiki_generation_managed: bool | None = None,
|
|
wiki_config_schema_version: int | None = None,
|
|
wiki_config_revision: int | None = None,
|
|
wiki_config_apply_status: str | None = None,
|
|
wiki_config_apply_error: str | None = None,
|
|
) -> dict[str, Any]: ...
|
|
|
|
@abstractmethod
|
|
async def list_visible(self, user_id: str, *, scope: LlmWikiScope = "all", is_admin: bool = False) -> list[dict[str, Any]]: ...
|
|
|
|
@abstractmethod
|
|
async def get_authorized(self, mapping_id: str, user_id: str, *, write: bool, is_admin: bool) -> dict[str, Any] | None: ...
|
|
|
|
@abstractmethod
|
|
async def get_by_weknora_id(self, weknora_id: str) -> dict[str, Any] | None: ...
|
|
|
|
@abstractmethod
|
|
async def update_mapping(
|
|
self,
|
|
mapping_id: str,
|
|
*,
|
|
name: str | None = None,
|
|
description: str | None = None,
|
|
test_questions: list[str] | None = None,
|
|
test_questions_set: bool = False,
|
|
retrieval_settings: dict[str, Any] | None = None,
|
|
wiki_index_enabled: bool | None = None,
|
|
external_search_enabled: bool | None = None,
|
|
wiki_parent_directories: list[str] | None | object = UNSET,
|
|
wiki_directory_paths: list[list[str]] | None | object = UNSET,
|
|
wiki_focus_instructions: str | None | object = UNSET,
|
|
wiki_content_instructions_raw: str | None | object = UNSET,
|
|
wiki_extraction_instructions_raw: str | None | object = UNSET,
|
|
wiki_generation_managed: bool | None | object = UNSET,
|
|
wiki_config_schema_version: int | None | object = UNSET,
|
|
wiki_config_revision: int | None | object = UNSET,
|
|
wiki_config_apply_status: str | None | object = UNSET,
|
|
wiki_config_apply_error: str | None | object = UNSET,
|
|
) -> dict[str, Any] | None: ...
|
|
|
|
@abstractmethod
|
|
async def adopt_conversation_deposit_mapping(
|
|
self,
|
|
mapping_id: str,
|
|
*,
|
|
name: str,
|
|
description: str,
|
|
) -> dict[str, Any] | None: ...
|
|
|
|
@abstractmethod
|
|
async def delete_mapping(self, mapping_id: str) -> bool: ...
|
|
|
|
@abstractmethod
|
|
async def request_publish(self, mapping_id: str, actor_user_id: str, *, is_admin: bool) -> dict[str, Any] | None: ...
|
|
|
|
@abstractmethod
|
|
async def review_publish(self, mapping_id: str, reviewer_user_id: str, *, approved: bool) -> dict[str, Any] | None: ...
|
|
|
|
@abstractmethod
|
|
async def unpublish(self, mapping_id: str, actor_user_id: str, *, is_admin: bool) -> dict[str, Any] | None: ...
|
|
|
|
@abstractmethod
|
|
async def list_publications(self, *, status: str | None = None) -> list[dict[str, Any]]: ...
|
|
|
|
@abstractmethod
|
|
async def get_conversation_deposit_mapping(self) -> dict[str, Any] | None: ...
|
|
|
|
@abstractmethod
|
|
async def get_conversation_deposit(
|
|
self,
|
|
*,
|
|
thread_id: str,
|
|
assistant_message_id: str,
|
|
) -> dict[str, Any] | None: ...
|
|
|
|
@abstractmethod
|
|
async def upsert_conversation_deposit(
|
|
self,
|
|
*,
|
|
thread_id: str,
|
|
human_message_id: str | None,
|
|
assistant_message_id: str,
|
|
knowledge_base_mapping_id: str,
|
|
weknora_knowledge_id: str | None,
|
|
title: str,
|
|
created_by_user_id: str | None,
|
|
) -> dict[str, Any]: ...
|
|
|
|
@abstractmethod
|
|
async def delete_conversation_deposits(
|
|
self,
|
|
*,
|
|
thread_id: str,
|
|
message_ids: list[str] | None = None,
|
|
) -> list[dict[str, Any]]: ...
|