forked from openai/chatkit-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathstore.py
More file actions
141 lines (109 loc) · 3.6 KB
/
Copy pathstore.py
File metadata and controls
141 lines (109 loc) · 3.6 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
import uuid
from abc import ABC, abstractmethod
from typing import Any, Generic, Literal
from typing_extensions import TypeVar
from .types import (
Attachment,
AttachmentCreateParams,
Page,
ThreadItem,
ThreadMetadata,
)
TContext = TypeVar("TContext", default=Any)
StoreItemType = Literal[
"thread", "message", "tool_call", "task", "workflow", "attachment"
]
_ID_PREFIXES: dict[StoreItemType, str] = {
"thread": "thr",
"message": "msg",
"tool_call": "tc",
"workflow": "wf",
"task": "tsk",
"attachment": "atc",
}
def default_generate_id(item_type: StoreItemType) -> str:
prefix = _ID_PREFIXES[item_type]
return f"{prefix}_{uuid.uuid4().hex[:8]}"
class NotFoundError(Exception):
pass
class AttachmentStore(ABC, Generic[TContext]):
@abstractmethod
async def delete_attachment(self, attachment_id: str, context: TContext) -> None:
pass
async def create_attachment(
self, input: AttachmentCreateParams, context: TContext
) -> Attachment:
raise NotImplementedError(
f"{type(self).__name__} must override create_attachment() to support two-phase file upload"
)
def generate_attachment_id(self, mime_type: str, context: TContext) -> str:
"""Return a new identifier for a file. Override this method to customize file ID generation."""
return default_generate_id("attachment")
class Store(ABC, Generic[TContext]):
def generate_thread_id(self, context: TContext) -> str:
"""Return a new identifier for a thread. Override this method to customize thread ID generation."""
return default_generate_id("thread")
def generate_item_id(
self, item_type: StoreItemType, thread: ThreadMetadata, context: TContext
) -> str:
"""Return a new identifier for a thread item. Override this method to customize item ID generation."""
return default_generate_id(item_type)
@abstractmethod
async def load_thread(self, thread_id: str, context: TContext) -> ThreadMetadata:
pass
@abstractmethod
async def save_thread(self, thread: ThreadMetadata, context: TContext) -> None:
pass
@abstractmethod
async def load_thread_items(
self,
thread_id: str,
after: str | None,
limit: int,
order: str,
context: TContext,
) -> Page[ThreadItem]:
pass
@abstractmethod
async def save_attachment(self, attachment: Attachment, context: TContext) -> None:
pass
@abstractmethod
async def load_attachment(
self, attachment_id: str, context: TContext
) -> Attachment:
pass
@abstractmethod
async def delete_attachment(self, attachment_id: str, context: TContext) -> None:
pass
@abstractmethod
async def load_threads(
self,
limit: int,
after: str | None,
order: str,
context: TContext,
) -> Page[ThreadMetadata]:
pass
@abstractmethod
async def add_thread_item(
self, thread_id: str, item: ThreadItem, context: TContext
) -> None:
pass
@abstractmethod
async def save_item(
self, thread_id: str, item: ThreadItem, context: TContext
) -> None:
pass
@abstractmethod
async def load_item(
self, thread_id: str, item_id: str, context: TContext
) -> ThreadItem:
pass
@abstractmethod
async def delete_thread(self, thread_id: str, context: TContext) -> None:
pass
@abstractmethod
async def delete_thread_item(
self, thread_id: str, item_id: str, context: TContext
) -> None:
pass