84 lines
2.9 KiB
Python
84 lines
2.9 KiB
Python
"""Persistence queries for user-owned assistant conversations."""
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from app.models.entities import Conversation, ConversationMessage
|
|
|
|
|
|
class ConversationRepository:
|
|
def __init__(self, session: AsyncSession) -> None:
|
|
self.session = session
|
|
|
|
async def list_for_user(
|
|
self,
|
|
user_id: int,
|
|
*,
|
|
limit: int = 50,
|
|
search: str | None = None,
|
|
) -> list[Conversation]:
|
|
filters = [Conversation.user_id == user_id]
|
|
if search:
|
|
filters.append(Conversation.title.ilike(f"%{search}%"))
|
|
statement = (
|
|
select(Conversation)
|
|
.where(*filters)
|
|
.options(selectinload(Conversation.messages))
|
|
.order_by(Conversation.updated_at.desc(), Conversation.id.desc())
|
|
.limit(limit)
|
|
)
|
|
return list(await self.session.scalars(statement))
|
|
|
|
async def get_page_for_user(
|
|
self,
|
|
conversation_id: int,
|
|
user_id: int,
|
|
*,
|
|
limit: int = 20,
|
|
offset: int = 0,
|
|
) -> tuple[Conversation, list[ConversationMessage], int] | None:
|
|
conversation = await self.session.scalar(
|
|
select(Conversation)
|
|
.where(
|
|
Conversation.id == conversation_id,
|
|
Conversation.user_id == user_id,
|
|
)
|
|
)
|
|
if conversation is None:
|
|
return None
|
|
total = await self.session.scalar(
|
|
select(func.count()).select_from(ConversationMessage).where(
|
|
ConversationMessage.conversation_id == conversation_id,
|
|
)
|
|
)
|
|
messages = list(
|
|
await self.session.scalars(
|
|
select(ConversationMessage)
|
|
.where(ConversationMessage.conversation_id == conversation_id)
|
|
.order_by(ConversationMessage.created_at.desc(), ConversationMessage.id.desc())
|
|
.offset(offset)
|
|
.limit(limit)
|
|
)
|
|
)
|
|
messages.reverse()
|
|
return conversation, messages, int(total or 0)
|
|
|
|
async def get_for_user(self, conversation_id: int, user_id: int) -> Conversation | None:
|
|
statement = (
|
|
select(Conversation)
|
|
.where(Conversation.id == conversation_id, Conversation.user_id == user_id)
|
|
.options(selectinload(Conversation.messages))
|
|
)
|
|
return await self.session.scalar(statement)
|
|
|
|
async def delete_for_user(self, conversation_id: int, user_id: int) -> bool:
|
|
"""Delete a conversation only when it belongs to the requesting user."""
|
|
|
|
conversation = await self.get_for_user(conversation_id, user_id)
|
|
if conversation is None:
|
|
return False
|
|
await self.session.delete(conversation)
|
|
await self.session.flush()
|
|
return True
|