synapse / backend /app /models.py
adwitiyashukla's picture
pushed Synapse, an agentic AI assistant with hybrid RAG
b2c1c67
Raw
History Blame Contribute Delete
4.49 kB
"""SQLAlchemy ORM models."""
import uuid
from datetime import datetime, timezone
from sqlalchemy import (
Boolean,
DateTime,
Float,
ForeignKey,
Integer,
LargeBinary,
String,
Text,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import Base
def utcnow() -> datetime:
return datetime.now(timezone.utc)
def new_uuid() -> str:
return uuid.uuid4().hex
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
username: Mapped[str] = mapped_column(String(80))
password_hash: Mapped[str] = mapped_column(String(255))
is_active: Mapped[bool] = mapped_column(Boolean, default=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
sessions: Mapped[list["ChatSession"]] = relationship(
back_populates="user", cascade="all, delete-orphan"
)
documents: Mapped[list["Document"]] = relationship(
back_populates="user", cascade="all, delete-orphan"
)
class ChatSession(Base):
__tablename__ = "chat_sessions"
id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_uuid)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True)
title: Mapped[str] = mapped_column(String(200), default="New chat")
model: Mapped[str | None] = mapped_column(String(80), nullable=True)
summary: Mapped[str | None] = mapped_column(Text, nullable=True)
summarized_until: Mapped[int] = mapped_column(Integer, default=0)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=utcnow, onupdate=utcnow
)
user: Mapped[User] = relationship(back_populates="sessions")
messages: Mapped[list["Message"]] = relationship(
back_populates="session",
cascade="all, delete-orphan",
order_by="Message.id",
)
class Message(Base):
__tablename__ = "messages"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
session_id: Mapped[str] = mapped_column(ForeignKey("chat_sessions.id"), index=True)
role: Mapped[str] = mapped_column(String(20)) # user | assistant
content: Mapped[str] = mapped_column(Text)
tool_calls_json: Mapped[str | None] = mapped_column(Text, nullable=True)
citations_json: Mapped[str | None] = mapped_column(Text, nullable=True)
model: Mapped[str | None] = mapped_column(String(80), nullable=True)
input_tokens: Mapped[int] = mapped_column(Integer, default=0)
output_tokens: Mapped[int] = mapped_column(Integer, default=0)
cost_usd: Mapped[float] = mapped_column(Float, default=0.0)
latency_ms: Mapped[int] = mapped_column(Integer, default=0)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
session: Mapped[ChatSession] = relationship(back_populates="messages")
class Document(Base):
__tablename__ = "documents"
id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_uuid)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True)
filename: Mapped[str] = mapped_column(String(255))
content_type: Mapped[str] = mapped_column(String(120))
size_bytes: Mapped[int] = mapped_column(Integer, default=0)
status: Mapped[str] = mapped_column(String(20), default="processing") # processing | ready | failed
error: Mapped[str | None] = mapped_column(Text, nullable=True)
chunk_count: Mapped[int] = mapped_column(Integer, default=0)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
user: Mapped[User] = relationship(back_populates="documents")
chunks: Mapped[list["Chunk"]] = relationship(
back_populates="document", cascade="all, delete-orphan"
)
class Chunk(Base):
__tablename__ = "chunks"
id: Mapped[str] = mapped_column(String(64), primary_key=True)
document_id: Mapped[str] = mapped_column(ForeignKey("documents.id"), index=True)
user_id: Mapped[int] = mapped_column(Integer, index=True)
chunk_index: Mapped[int] = mapped_column(Integer)
text: Mapped[str] = mapped_column(Text)
embedding: Mapped[bytes | None] = mapped_column(LargeBinary, nullable=True)
document: Mapped[Document] = relationship(back_populates="chunks")