conversation_variable_updater.py

from sqlalchemy import select
from sqlalchemy.orm import Session

from core.variables.variables import Variable
from extensions.ext_database import db
from models import ConversationVariable


class ConversationVariableNotFoundError(Exception):
    pass


class ConversationVariableUpdaterImpl:
    def update(self, conversation_id: str, variable: Variable) -> None:
        stmt = select(ConversationVariable).where(
            ConversationVariable.id == variable.id, ConversationVariable.conversation_id == conversation_id
        )
        session = Session(db.engine)
        row = session.scalar(stmt)
        if not row:
            raise ConversationVariableNotFoundError("conversation variable not found in the database")
        row.data = variable.model_dump_json()
        session.commit()

    def flush(self) -> None:
        pass


def conversation_variable_updater_factory() -> ConversationVariableUpdaterImpl:
    return ConversationVariableUpdaterImpl()