| from sqlmodel import Field, SQLModel, Relationship, Column, JSON |
| from typing import Optional, Dict, Any, TYPE_CHECKING |
| from datetime import datetime |
| from enum import Enum |
|
|
| if TYPE_CHECKING: |
| from .conversation import Conversation |
|
|
|
|
| class MessageRole(str, Enum): |
| """Enum representing the role of a message sender.""" |
| USER = "user" |
| ASSISTANT = "assistant" |
|
|
|
|
| class Message(SQLModel, table=True): |
| """Message model representing a single exchange in a conversation.""" |
|
|
| __tablename__ = "messages" |
| __table_args__ = ( |
| {"sqlite_autoincrement": True}, |
| ) |
|
|
| id: Optional[int] = Field(default=None, primary_key=True) |
| conversation_id: int = Field(foreign_key="conversations.id", index=True) |
| role: MessageRole = Field(sa_column_kwargs={"nullable": False}) |
| content: str = Field(sa_column_kwargs={"nullable": False}) |
| tool_calls: Optional[Dict[str, Any]] = Field( |
| default=None, |
| sa_column=Column(JSON) |
| ) |
| sequence_number: int = Field(index=True) |
| created_at: datetime = Field(default_factory=datetime.utcnow, index=True) |
| deleted_at: Optional[datetime] = Field(default=None, index=True) |
|
|
| |
| conversation: "Conversation" = Relationship(back_populates="messages") |
|
|