Skip to content

store

AttachmentStore

Bases: ABC, Generic[TContext]

Source code in chatkit/store.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
class AttachmentStore(ABC, Generic[TContext]):
    @abstractmethod
    async def delete_attachment(self, attachment_id: str, context: TContext) -> None:
        """Delete an attachment by id."""
        pass

    async def create_attachment(
        self, input: AttachmentCreateParams, context: TContext
    ) -> Attachment:
        """Create an attachment record from upload metadata."""
        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")

delete_attachment abstractmethod async 

delete_attachment(
    attachment_id: str, context: TContext
) -> None

Delete an attachment by id.

Source code in chatkit/store.py 
49
50
51
52
@abstractmethod
async def delete_attachment(self, attachment_id: str, context: TContext) -> None:
    """Delete an attachment by id."""
    pass

create_attachment async 

create_attachment(
    input: AttachmentCreateParams, context: TContext
) -> Attachment

Create an attachment record from upload metadata.

Source code in chatkit/store.py 
54
55
56
57
58
59
60
async def create_attachment(
    self, input: AttachmentCreateParams, context: TContext
) -> Attachment:
    """Create an attachment record from upload metadata."""
    raise NotImplementedError(
        f"{type(self).__name__} must override create_attachment() to support two-phase file upload"
    )

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 
62
63
64
65
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 
 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
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
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:
        """Load a thread's metadata by id."""
        pass

    @abstractmethod
    async def save_thread(self, thread: ThreadMetadata, context: TContext) -> None:
        """Persist thread metadata (title, status, etc.)."""
        pass

    @abstractmethod
    async def load_thread_items(
        self,
        thread_id: str,
        after: str | None,
        limit: int,
        order: str,
        context: TContext,
    ) -> Page[ThreadItem]:
        """Load a page of thread items with pagination controls."""
        pass

    @abstractmethod
    async def save_attachment(self, attachment: Attachment, context: TContext) -> None:
        """Persist attachment metadata."""
        pass

    @abstractmethod
    async def load_attachment(
        self, attachment_id: str, context: TContext
    ) -> Attachment:
        """Load attachment metadata by id."""
        pass

    @abstractmethod
    async def delete_attachment(self, attachment_id: str, context: TContext) -> None:
        """Delete attachment metadata by id."""
        pass

    @abstractmethod
    async def load_threads(
        self,
        limit: int,
        after: str | None,
        order: str,
        context: TContext,
    ) -> Page[ThreadMetadata]:
        """Load a page of threads with pagination controls."""
        pass

    @abstractmethod
    async def add_thread_item(
        self, thread_id: str, item: ThreadItem, context: TContext
    ) -> None:
        """Persist a newly created thread item."""
        pass

    @abstractmethod
    async def save_item(
        self, thread_id: str, item: ThreadItem, context: TContext
    ) -> None:
        """Upsert a thread item by id."""
        pass

    @abstractmethod
    async def load_item(
        self, thread_id: str, item_id: str, context: TContext
    ) -> ThreadItem:
        """Load a thread item by id."""
        pass

    @abstractmethod
    async def delete_thread(self, thread_id: str, context: TContext) -> None:
        """Delete a thread and its items."""
        pass

    @abstractmethod
    async def delete_thread_item(
        self, thread_id: str, item_id: str, context: TContext
    ) -> None:
        """Delete a thread item by id."""
        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 
69
70
71
72
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 
74
75
76
77
78
79
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)

load_thread abstractmethod async 

load_thread(
    thread_id: str, context: TContext
) -> ThreadMetadata

Load a thread's metadata by id.

Source code in chatkit/store.py 
81
82
83
84
@abstractmethod
async def load_thread(self, thread_id: str, context: TContext) -> ThreadMetadata:
    """Load a thread's metadata by id."""
    pass

save_thread abstractmethod async 

save_thread(
    thread: ThreadMetadata, context: TContext
) -> None

Persist thread metadata (title, status, etc.).

Source code in chatkit/store.py 
86
87
88
89
@abstractmethod
async def save_thread(self, thread: ThreadMetadata, context: TContext) -> None:
    """Persist thread metadata (title, status, etc.)."""
    pass

load_thread_items abstractmethod async 

load_thread_items(
    thread_id: str,
    after: str | None,
    limit: int,
    order: str,
    context: TContext,
) -> Page[ThreadItem]

Load a page of thread items with pagination controls.

Source code in chatkit/store.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
@abstractmethod
async def load_thread_items(
    self,
    thread_id: str,
    after: str | None,
    limit: int,
    order: str,
    context: TContext,
) -> Page[ThreadItem]:
    """Load a page of thread items with pagination controls."""
    pass

save_attachment abstractmethod async 

save_attachment(
    attachment: Attachment, context: TContext
) -> None

Persist attachment metadata.

Source code in chatkit/store.py 
103
104
105
106
@abstractmethod
async def save_attachment(self, attachment: Attachment, context: TContext) -> None:
    """Persist attachment metadata."""
    pass

load_attachment abstractmethod async 

load_attachment(
    attachment_id: str, context: TContext
) -> Attachment

Load attachment metadata by id.

Source code in chatkit/store.py 
108
109
110
111
112
113
@abstractmethod
async def load_attachment(
    self, attachment_id: str, context: TContext
) -> Attachment:
    """Load attachment metadata by id."""
    pass

delete_attachment abstractmethod async 

delete_attachment(
    attachment_id: str, context: TContext
) -> None

Delete attachment metadata by id.

Source code in chatkit/store.py 
115
116
117
118
@abstractmethod
async def delete_attachment(self, attachment_id: str, context: TContext) -> None:
    """Delete attachment metadata by id."""
    pass

load_threads abstractmethod async 

load_threads(
    limit: int,
    after: str | None,
    order: str,
    context: TContext,
) -> Page[ThreadMetadata]

Load a page of threads with pagination controls.

Source code in chatkit/store.py 
120
121
122
123
124
125
126
127
128
129
@abstractmethod
async def load_threads(
    self,
    limit: int,
    after: str | None,
    order: str,
    context: TContext,
) -> Page[ThreadMetadata]:
    """Load a page of threads with pagination controls."""
    pass

add_thread_item abstractmethod async 

add_thread_item(
    thread_id: str, item: ThreadItem, context: TContext
) -> None

Persist a newly created thread item.

Source code in chatkit/store.py 
131
132
133
134
135
136
@abstractmethod
async def add_thread_item(
    self, thread_id: str, item: ThreadItem, context: TContext
) -> None:
    """Persist a newly created thread item."""
    pass

save_item abstractmethod async 

save_item(
    thread_id: str, item: ThreadItem, context: TContext
) -> None

Upsert a thread item by id.

Source code in chatkit/store.py 
138
139
140
141
142
143
@abstractmethod
async def save_item(
    self, thread_id: str, item: ThreadItem, context: TContext
) -> None:
    """Upsert a thread item by id."""
    pass

load_item abstractmethod async 

load_item(
    thread_id: str, item_id: str, context: TContext
) -> ThreadItem

Load a thread item by id.

Source code in chatkit/store.py 
145
146
147
148
149
150
@abstractmethod
async def load_item(
    self, thread_id: str, item_id: str, context: TContext
) -> ThreadItem:
    """Load a thread item by id."""
    pass

delete_thread abstractmethod async 

delete_thread(thread_id: str, context: TContext) -> None

Delete a thread and its items.

Source code in chatkit/store.py 
152
153
154
155
@abstractmethod
async def delete_thread(self, thread_id: str, context: TContext) -> None:
    """Delete a thread and its items."""
    pass

delete_thread_item abstractmethod async 

delete_thread_item(
    thread_id: str, item_id: str, context: TContext
) -> None

Delete a thread item by id.

Source code in chatkit/store.py 
157
158
159
160
161
162
@abstractmethod
async def delete_thread_item(
    self, thread_id: str, item_id: str, context: TContext
) -> None:
    """Delete a thread item by id."""
    pass