Skip to content

store

AttachmentStore

Bases: ABC, Generic[TContext]

Source code in chatkit/store.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
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")

generate_attachment_id 

generate_attachment_id(
    mime_type: str, context: TContext
) -> str

Return a new identifier for a file. Override this method to customize file ID generation.

Source code in chatkit/store.py 
53
54
55
56
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")

Store

Bases: ABC, Generic[TContext]

Source code in chatkit/store.py 
 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
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

generate_thread_id 

generate_thread_id(context: TContext) -> str

Return a new identifier for a thread. Override this method to customize thread ID generation.

Source code in chatkit/store.py 
60
61
62
63
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")

generate_item_id 

generate_item_id(
    item_type: StoreItemType,
    thread: ThreadMetadata,
    context: TContext,
) -> str

Return a new identifier for a thread item. Override this method to customize item ID generation.

Source code in chatkit/store.py 
65
66
67
68
69
70
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)