mirror of https://github.com/langgenius/dify.git
29 lines
984 B
Python
29 lines
984 B
Python
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
from core.variables.variables import Variable
|
|
from models import ConversationVariable
|
|
|
|
|
|
class ConversationVariableNotFoundError(Exception):
|
|
pass
|
|
|
|
|
|
class ConversationVariableUpdater:
|
|
def __init__(self, session_maker: sessionmaker[Session]) -> None:
|
|
self._session_maker: sessionmaker[Session] = session_maker
|
|
|
|
def update(self, conversation_id: str, variable: Variable) -> None:
|
|
stmt = select(ConversationVariable).where(
|
|
ConversationVariable.id == variable.id, ConversationVariable.conversation_id == conversation_id
|
|
)
|
|
with self._session_maker() as session:
|
|
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
|