api/models/web.py

Rama main · réplica de lectura del repositorio original.

from datetime import datetime
from uuid import uuid4

import sqlalchemy as sa
from sqlalchemy import DateTime, func, select
from sqlalchemy.orm import Mapped, Session, mapped_column

from .base import TypeBase
from .enums import CreatorUserRole
from .model import Message
from .types import EnumText, StringUUID


class SavedMessage(TypeBase):
    __tablename__ = "saved_messages"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="saved_message_pkey"),
        sa.Index("saved_message_message_idx", "app_id", "message_id", "created_by_role", "created_by"),
        sa.Index("saved_message_message_id_idx", "message_id"),
    )

    id: Mapped[str] = mapped_column(
        StringUUID, insert_default=lambda: str(uuid4()), default_factory=lambda: str(uuid4()), init=False
    )
    app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    message_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    created_by_role: Mapped[CreatorUserRole] = mapped_column(
        EnumText(CreatorUserRole, length=255), nullable=False, server_default=sa.text("'end_user'")
    )
    created_by: Mapped[str] = mapped_column(StringUUID, nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime,
        nullable=False,
        server_default=func.current_timestamp(),
        init=False,
    )

    def message(self, session: Session) -> Message | None:
        return session.scalar(select(Message).where(Message.id == self.message_id))


class PinnedConversation(TypeBase):
    __tablename__ = "pinned_conversations"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="pinned_conversation_pkey"),
        sa.Index("pinned_conversation_conversation_idx", "app_id", "conversation_id", "created_by_role", "created_by"),
    )

    id: Mapped[str] = mapped_column(
        StringUUID, insert_default=lambda: str(uuid4()), default_factory=lambda: str(uuid4()), init=False
    )
    app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    conversation_id: Mapped[str] = mapped_column(StringUUID)
    created_by_role: Mapped[CreatorUserRole] = mapped_column(
        EnumText(CreatorUserRole, length=255),
        nullable=False,
        server_default=sa.text("'end_user'"),
    )
    created_by: Mapped[str] = mapped_column(StringUUID, nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime,
        nullable=False,
        server_default=func.current_timestamp(),
        init=False,
    )