Underground-Digital/Workflow-Engine
0
1import json2from collections.abc import Mapping, Sequence3from datetime import datetime, timezone4from enum import Enum5from typing import Any, Optional, Union6 7import sqlalchemy as sa8from sqlalchemy import func9from sqlalchemy.orm import Mapped, mapped_column10 11import contexts12from constants import HIDDEN_VALUE13from core.helper import encrypter14from core.variables import SecretVariable, Variable15from extensions.ext_database import db16from factories import variable_factory17from libs import helper18from models.enums import CreatedByRole19 20from .account import Account21from .types import StringUUID22 23 24class WorkflowType(Enum):25 """26 Workflow Type Enum27 """28 29 WORKFLOW = "workflow"30 CHAT = "chat"31 32 @classmethod33 def value_of(cls, value: str) -> "WorkflowType":34 """35 Get value of given mode.36 37 :param value: mode value38 :return: mode39 """40 for mode in cls:41 if mode.value == value:42 return mode43 raise ValueError(f"invalid workflow type value {value}")44 45 @classmethod46 def from_app_mode(cls, app_mode: Union[str, "AppMode"]) -> "WorkflowType":47 """48 Get workflow type from app mode.49 50 :param app_mode: app mode51 :return: workflow type52 """53 from models.model import AppMode54 55 app_mode = app_mode if isinstance(app_mode, AppMode) else AppMode.value_of(app_mode)56 return cls.WORKFLOW if app_mode == AppMode.WORKFLOW else cls.CHAT57 58 59class Workflow(db.Model):60 """61 Workflow, for `Workflow App` and `Chat App workflow mode`.62 63 Attributes:64 65 - id (uuid) Workflow ID, pk66 - tenant_id (uuid) Workspace ID67 - app_id (uuid) App ID68 - type (string) Workflow type69 70 `workflow` for `Workflow App`71 72 `chat` for `Chat App workflow mode`73 74 - version (string) Version75 76 `draft` for draft version (only one for each app), other for version number (redundant)77 78 - graph (text) Workflow canvas configuration (JSON)79 80 The entire canvas configuration JSON, including Node, Edge, and other configurations81 82 - nodes (array[object]) Node list, see Node Schema83 84 - edges (array[object]) Edge list, see Edge Schema85 86 - created_by (uuid) Creator ID87 - created_at (timestamp) Creation time88 - updated_by (uuid) `optional` Last updater ID89 - updated_at (timestamp) `optional` Last update time90 """91 92 __tablename__ = "workflows"93 __table_args__ = (94 db.PrimaryKeyConstraint("id", name="workflow_pkey"),95 db.Index("workflow_version_idx", "tenant_id", "app_id", "version"),96 )97 98 id: Mapped[str] = mapped_column(StringUUID, server_default=db.text("uuid_generate_v4()"))99 tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)100 app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)101 type: Mapped[str] = mapped_column(db.String(255), nullable=False)102 version: Mapped[str] = mapped_column(db.String(255), nullable=False)103 graph: Mapped[str] = mapped_column(sa.Text)104 _features: Mapped[str] = mapped_column("features", sa.TEXT)105 created_by: Mapped[str] = mapped_column(StringUUID, nullable=False)106 created_at: Mapped[datetime] = mapped_column(107 db.DateTime, nullable=False, server_default=db.text("CURRENT_TIMESTAMP(0)")108 )109 updated_by: Mapped[Optional[str]] = mapped_column(StringUUID)110 updated_at: Mapped[datetime] = mapped_column(111 sa.DateTime, nullable=False, default=datetime.now(tz=timezone.utc), server_onupdate=func.current_timestamp()112 )113 _environment_variables: Mapped[str] = mapped_column(114 "environment_variables", db.Text, nullable=False, server_default="{}"115 )116 _conversation_variables: Mapped[str] = mapped_column(117 "conversation_variables", db.Text, nullable=False, server_default="{}"118 )119 120 def __init__(121 self,122 *,123 tenant_id: str,124 app_id: str,125 type: str,126 version: str,127 graph: str,128 features: str,129 created_by: str,130 environment_variables: Sequence[Variable],131 conversation_variables: Sequence[Variable],132 ):133 self.tenant_id = tenant_id134 self.app_id = app_id135 self.type = type136 self.version = version137 self.graph = graph138 self.features = features139 self.created_by = created_by140 self.environment_variables = environment_variables or []141 self.conversation_variables = conversation_variables or []142 143 @property144 def created_by_account(self):145 return db.session.get(Account, self.created_by)146 147 @property148 def updated_by_account(self):149 return db.session.get(Account, self.updated_by) if self.updated_by else None150 151 @property152 def graph_dict(self) -> Mapping[str, Any]:153 return json.loads(self.graph) if self.graph else {}154 155 @property156 def features(self) -> str:157 """158 Convert old features structure to new features structure.159 """160 if not self._features:161 return self._features162 163 features = json.loads(self._features)164 if features.get("file_upload", {}).get("image", {}).get("enabled", False):165 image_enabled = True166 image_number_limits = int(features["file_upload"]["image"].get("number_limits", 1))167 image_transfer_methods = features["file_upload"]["image"].get(168 "transfer_methods", ["remote_url", "local_file"]169 )170 features["file_upload"]["enabled"] = image_enabled171 features["file_upload"]["number_limits"] = image_number_limits172 features["file_upload"]["allowed_upload_methods"] = image_transfer_methods173 features["file_upload"]["allowed_file_types"] = ["image"]174 features["file_upload"]["allowed_extensions"] = []175 del features["file_upload"]["image"]176 self._features = json.dumps(features)177 return self._features178 179 @features.setter180 def features(self, value: str) -> None:181 self._features = value182 183 @property184 def features_dict(self) -> Mapping[str, Any]:185 return json.loads(self.features) if self.features else {}186 187 def user_input_form(self, to_old_structure: bool = False) -> list:188 # get start node from graph189 if not self.graph:190 return []191 192 graph_dict = self.graph_dict193 if "nodes" not in graph_dict:194 return []195 196 start_node = next((node for node in graph_dict["nodes"] if node["data"]["type"] == "start"), None)197 if not start_node:198 return []199 200 # get user_input_form from start node201 variables = start_node.get("data", {}).get("variables", [])202 203 if to_old_structure:204 old_structure_variables = []205 for variable in variables:206 old_structure_variables.append({variable["type"]: variable})207 208 return old_structure_variables209 210 return variables211 212 @property213 def unique_hash(self) -> str:214 """215 Get hash of workflow.216 217 :return: hash218 """219 entity = {"graph": self.graph_dict, "features": self.features_dict}220 221 return helper.generate_text_hash(json.dumps(entity, sort_keys=True))222 223 @property224 def tool_published(self) -> bool:225 from models.tools import WorkflowToolProvider226 227 return (228 db.session.query(WorkflowToolProvider).filter(WorkflowToolProvider.app_id == self.app_id).first()229 is not None230 )231 232 @property233 def environment_variables(self) -> Sequence[Variable]:234 # TODO: find some way to init `self._environment_variables` when instance created.235 if self._environment_variables is None:236 self._environment_variables = "{}"237 238 tenant_id = contexts.tenant_id.get()239 240 environment_variables_dict: dict[str, Any] = json.loads(self._environment_variables)241 results = [variable_factory.build_variable_from_mapping(v) for v in environment_variables_dict.values()]242 243 # decrypt secret variables value244 decrypt_func = (245 lambda var: var.model_copy(update={"value": encrypter.decrypt_token(tenant_id=tenant_id, token=var.value)})246 if isinstance(var, SecretVariable)247 else var248 )249 results = list(map(decrypt_func, results))250 return results251 252 @environment_variables.setter253 def environment_variables(self, value: Sequence[Variable]):254 if not value:255 self._environment_variables = "{}"256 return257 258 tenant_id = contexts.tenant_id.get()259 260 value = list(value)261 if any(var for var in value if not var.id):262 raise ValueError("environment variable require a unique id")263 264 # Compare inputs and origin variables,265 # if the value is HIDDEN_VALUE, use the origin variable value (only update `name`).266 origin_variables_dictionary = {var.id: var for var in self.environment_variables}267 for i, variable in enumerate(value):268 if variable.id in origin_variables_dictionary and variable.value == HIDDEN_VALUE:269 value[i] = origin_variables_dictionary[variable.id].model_copy(update={"name": variable.name})270 271 # encrypt secret variables value272 encrypt_func = (273 lambda var: var.model_copy(update={"value": encrypter.encrypt_token(tenant_id=tenant_id, token=var.value)})274 if isinstance(var, SecretVariable)275 else var276 )277 encrypted_vars = list(map(encrypt_func, value))278 environment_variables_json = json.dumps(279 {var.name: var.model_dump() for var in encrypted_vars},280 ensure_ascii=False,281 )282 self._environment_variables = environment_variables_json283 284 def to_dict(self, *, include_secret: bool = False) -> Mapping[str, Any]:285 environment_variables = list(self.environment_variables)286 environment_variables = [287 v if not isinstance(v, SecretVariable) or include_secret else v.model_copy(update={"value": ""})288 for v in environment_variables289 ]290 291 result = {292 "graph": self.graph_dict,293 "features": self.features_dict,294 "environment_variables": [var.model_dump(mode="json") for var in environment_variables],295 "conversation_variables": [var.model_dump(mode="json") for var in self.conversation_variables],296 }297 return result298 299 @property300 def conversation_variables(self) -> Sequence[Variable]:301 # TODO: find some way to init `self._conversation_variables` when instance created.302 if self._conversation_variables is None:303 self._conversation_variables = "{}"304 305 variables_dict: dict[str, Any] = json.loads(self._conversation_variables)306 results = [variable_factory.build_variable_from_mapping(v) for v in variables_dict.values()]307 return results308 309 @conversation_variables.setter310 def conversation_variables(self, value: Sequence[Variable]) -> None:311 self._conversation_variables = json.dumps(312 {var.name: var.model_dump() for var in value},313 ensure_ascii=False,314 )315 316 317class WorkflowRunStatus(Enum):318 """319 Workflow Run Status Enum320 """321 322 RUNNING = "running"323 SUCCEEDED = "succeeded"324 FAILED = "failed"325 STOPPED = "stopped"326 327 @classmethod328 def value_of(cls, value: str) -> "WorkflowRunStatus":329 """330 Get value of given mode.331 332 :param value: mode value333 :return: mode334 """335 for mode in cls:336 if mode.value == value:337 return mode338 raise ValueError(f"invalid workflow run status value {value}")339 340 341class WorkflowRun(db.Model):342 """343 Workflow Run344 345 Attributes:346 347 - id (uuid) Run ID348 - tenant_id (uuid) Workspace ID349 - app_id (uuid) App ID350 - sequence_number (int) Auto-increment sequence number, incremented within the App, starting from 1351 - workflow_id (uuid) Workflow ID352 - type (string) Workflow type353 - triggered_from (string) Trigger source354 355 `debugging` for canvas debugging356 357 `app-run` for (published) app execution358 359 - version (string) Version360 - graph (text) Workflow canvas configuration (JSON)361 - inputs (text) Input parameters362 - status (string) Execution status, `running` / `succeeded` / `failed` / `stopped`363 - outputs (text) `optional` Output content364 - error (string) `optional` Error reason365 - elapsed_time (float) `optional` Time consumption (s)366 - total_tokens (int) `optional` Total tokens used367 - total_steps (int) Total steps (redundant), default 0368 - created_by_role (string) Creator role369 370 - `account` Console account371 372 - `end_user` End user373 374 - created_by (uuid) Runner ID375 - created_at (timestamp) Run time376 - finished_at (timestamp) End time377 """378 379 __tablename__ = "workflow_runs"380 __table_args__ = (381 db.PrimaryKeyConstraint("id", name="workflow_run_pkey"),382 db.Index("workflow_run_triggerd_from_idx", "tenant_id", "app_id", "triggered_from"),383 db.Index("workflow_run_tenant_app_sequence_idx", "tenant_id", "app_id", "sequence_number"),384 )385 386 id = db.Column(StringUUID, server_default=db.text("uuid_generate_v4()"))387 tenant_id = db.Column(StringUUID, nullable=False)388 app_id = db.Column(StringUUID, nullable=False)389 sequence_number = db.Column(db.Integer, nullable=False)390 workflow_id = db.Column(StringUUID, nullable=False)391 type = db.Column(db.String(255), nullable=False)392 triggered_from = db.Column(db.String(255), nullable=False)393 version = db.Column(db.String(255), nullable=False)394 graph = db.Column(db.Text)395 inputs = db.Column(db.Text)396 status = db.Column(db.String(255), nullable=False)397 outputs: Mapped[str] = db.Column(db.Text)398 error = db.Column(db.Text)399 elapsed_time = db.Column(db.Float, nullable=False, server_default=db.text("0"))400 total_tokens = db.Column(db.Integer, nullable=False, server_default=db.text("0"))401 total_steps = db.Column(db.Integer, server_default=db.text("0"))402 created_by_role = db.Column(db.String(255), nullable=False)403 created_by = db.Column(StringUUID, nullable=False)404 created_at = db.Column(db.DateTime, nullable=False, server_default=db.text("CURRENT_TIMESTAMP(0)"))405 finished_at = db.Column(db.DateTime)406 407 @property408 def created_by_account(self):409 created_by_role = CreatedByRole(self.created_by_role)410 return db.session.get(Account, self.created_by) if created_by_role == CreatedByRole.ACCOUNT else None411 412 @property413 def created_by_end_user(self):414 from models.model import EndUser415 416 created_by_role = CreatedByRole(self.created_by_role)417 return db.session.get(EndUser, self.created_by) if created_by_role == CreatedByRole.END_USER else None418 419 @property420 def graph_dict(self):421 return json.loads(self.graph) if self.graph else {}422 423 @property424 def inputs_dict(self) -> Mapping[str, Any]:425 return json.loads(self.inputs) if self.inputs else {}426 427 @property428 def outputs_dict(self) -> Mapping[str, Any]:429 return json.loads(self.outputs) if self.outputs else {}430 431 @property432 def message(self) -> Optional["Message"]:433 from models.model import Message434 435 return (436 db.session.query(Message).filter(Message.app_id == self.app_id, Message.workflow_run_id == self.id).first()437 )438 439 @property440 def workflow(self):441 return db.session.query(Workflow).filter(Workflow.id == self.workflow_id).first()442 443 def to_dict(self):444 return {445 "id": self.id,446 "tenant_id": self.tenant_id,447 "app_id": self.app_id,448 "sequence_number": self.sequence_number,449 "workflow_id": self.workflow_id,450 "type": self.type,451 "triggered_from": self.triggered_from,452 "version": self.version,453 "graph": self.graph_dict,454 "inputs": self.inputs_dict,455 "status": self.status,456 "outputs": self.outputs_dict,457 "error": self.error,458 "elapsed_time": self.elapsed_time,459 "total_tokens": self.total_tokens,460 "total_steps": self.total_steps,461 "created_by_role": self.created_by_role,462 "created_by": self.created_by,463 "created_at": self.created_at,464 "finished_at": self.finished_at,465 }466 467 @classmethod468 def from_dict(cls, data: dict) -> "WorkflowRun":469 return cls(470 id=data.get("id"),471 tenant_id=data.get("tenant_id"),472 app_id=data.get("app_id"),473 sequence_number=data.get("sequence_number"),474 workflow_id=data.get("workflow_id"),475 type=data.get("type"),476 triggered_from=data.get("triggered_from"),477 version=data.get("version"),478 graph=json.dumps(data.get("graph")),479 inputs=json.dumps(data.get("inputs")),480 status=data.get("status"),481 outputs=json.dumps(data.get("outputs")),482 error=data.get("error"),483 elapsed_time=data.get("elapsed_time"),484 total_tokens=data.get("total_tokens"),485 total_steps=data.get("total_steps"),486 created_by_role=data.get("created_by_role"),487 created_by=data.get("created_by"),488 created_at=data.get("created_at"),489 finished_at=data.get("finished_at"),490 )491 492 493class WorkflowNodeExecutionTriggeredFrom(Enum):494 """495 Workflow Node Execution Triggered From Enum496 """497 498 SINGLE_STEP = "single-step"499 WORKFLOW_RUN = "workflow-run"500 501 @classmethod502 def value_of(cls, value: str) -> "WorkflowNodeExecutionTriggeredFrom":503 """504 Get value of given mode.505 506 :param value: mode value507 :return: mode508 """509 for mode in cls:510 if mode.value == value:511 return mode512 raise ValueError(f"invalid workflow node execution triggered from value {value}")513 514 515class WorkflowNodeExecutionStatus(Enum):516 """517 Workflow Node Execution Status Enum518 """519 520 RUNNING = "running"521 SUCCEEDED = "succeeded"522 FAILED = "failed"523 524 @classmethod525 def value_of(cls, value: str) -> "WorkflowNodeExecutionStatus":526 """527 Get value of given mode.528 529 :param value: mode value530 :return: mode531 """532 for mode in cls:533 if mode.value == value:534 return mode535 raise ValueError(f"invalid workflow node execution status value {value}")536 537 538class WorkflowNodeExecution(db.Model):539 """540 Workflow Node Execution541 542 - id (uuid) Execution ID543 - tenant_id (uuid) Workspace ID544 - app_id (uuid) App ID545 - workflow_id (uuid) Workflow ID546 - triggered_from (string) Trigger source547 548 `single-step` for single-step debugging549 550 `workflow-run` for workflow execution (debugging / user execution)551 552 - workflow_run_id (uuid) `optional` Workflow run ID553 554 Null for single-step debugging.555 556 - index (int) Execution sequence number, used for displaying Tracing Node order557 - predecessor_node_id (string) `optional` Predecessor node ID, used for displaying execution path558 - node_id (string) Node ID559 - node_type (string) Node type, such as `start`560 - title (string) Node title561 - inputs (json) All predecessor node variable content used in the node562 - process_data (json) Node process data563 - outputs (json) `optional` Node output variables564 - status (string) Execution status, `running` / `succeeded` / `failed`565 - error (string) `optional` Error reason566 - elapsed_time (float) `optional` Time consumption (s)567 - execution_metadata (text) Metadata568 569 - total_tokens (int) `optional` Total tokens used570 571 - total_price (decimal) `optional` Total cost572 573 - currency (string) `optional` Currency, such as USD / RMB574 575 - created_at (timestamp) Run time576 - created_by_role (string) Creator role577 578 - `account` Console account579 580 - `end_user` End user581 582 - created_by (uuid) Runner ID583 - finished_at (timestamp) End time584 """585 586 __tablename__ = "workflow_node_executions"587 __table_args__ = (588 db.PrimaryKeyConstraint("id", name="workflow_node_execution_pkey"),589 db.Index(590 "workflow_node_execution_workflow_run_idx",591 "tenant_id",592 "app_id",593 "workflow_id",594 "triggered_from",595 "workflow_run_id",596 ),597 db.Index(598 "workflow_node_execution_node_run_idx", "tenant_id", "app_id", "workflow_id", "triggered_from", "node_id"599 ),600 db.Index(601 "workflow_node_execution_id_idx",602 "tenant_id",603 "app_id",604 "workflow_id",605 "triggered_from",606 "node_execution_id",607 ),608 )609 610 id = db.Column(StringUUID, server_default=db.text("uuid_generate_v4()"))611 tenant_id = db.Column(StringUUID, nullable=False)612 app_id = db.Column(StringUUID, nullable=False)613 workflow_id = db.Column(StringUUID, nullable=False)614 triggered_from = db.Column(db.String(255), nullable=False)615 workflow_run_id = db.Column(StringUUID)616 index = db.Column(db.Integer, nullable=False)617 predecessor_node_id = db.Column(db.String(255))618 node_execution_id = db.Column(db.String(255), nullable=True)619 node_id = db.Column(db.String(255), nullable=False)620 node_type = db.Column(db.String(255), nullable=False)621 title = db.Column(db.String(255), nullable=False)622 inputs = db.Column(db.Text)623 process_data = db.Column(db.Text)624 outputs = db.Column(db.Text)625 status = db.Column(db.String(255), nullable=False)626 error = db.Column(db.Text)627 elapsed_time = db.Column(db.Float, nullable=False, server_default=db.text("0"))628 execution_metadata = db.Column(db.Text)629 created_at = db.Column(db.DateTime, nullable=False, server_default=db.text("CURRENT_TIMESTAMP(0)"))630 created_by_role = db.Column(db.String(255), nullable=False)631 created_by = db.Column(StringUUID, nullable=False)632 finished_at = db.Column(db.DateTime)633 634 @property635 def created_by_account(self):636 created_by_role = CreatedByRole(self.created_by_role)637 return db.session.get(Account, self.created_by) if created_by_role == CreatedByRole.ACCOUNT else None638 639 @property640 def created_by_end_user(self):641 from models.model import EndUser642 643 created_by_role = CreatedByRole(self.created_by_role)644 return db.session.get(EndUser, self.created_by) if created_by_role == CreatedByRole.END_USER else None645 646 @property647 def inputs_dict(self):648 return json.loads(self.inputs) if self.inputs else None649 650 @property651 def outputs_dict(self):652 return json.loads(self.outputs) if self.outputs else None653 654 @property655 def process_data_dict(self):656 return json.loads(self.process_data) if self.process_data else None657 658 @property659 def execution_metadata_dict(self):660 return json.loads(self.execution_metadata) if self.execution_metadata else None661 662 @property663 def extras(self):664 from core.tools.tool_manager import ToolManager665 666 extras = {}667 if self.execution_metadata_dict:668 from core.workflow.nodes import NodeType669 670 if self.node_type == NodeType.TOOL.value and "tool_info" in self.execution_metadata_dict:671 tool_info = self.execution_metadata_dict["tool_info"]672 extras["icon"] = ToolManager.get_tool_icon(673 tenant_id=self.tenant_id,674 provider_type=tool_info["provider_type"],675 provider_id=tool_info["provider_id"],676 )677 678 return extras679 680 681class WorkflowAppLogCreatedFrom(Enum):682 """683 Workflow App Log Created From Enum684 """685 686 SERVICE_API = "service-api"687 WEB_APP = "web-app"688 INSTALLED_APP = "installed-app"689 690 @classmethod691 def value_of(cls, value: str) -> "WorkflowAppLogCreatedFrom":692 """693 Get value of given mode.694 695 :param value: mode value696 :return: mode697 """698 for mode in cls:699 if mode.value == value:700 return mode701 raise ValueError(f"invalid workflow app log created from value {value}")702 703 704class WorkflowAppLog(db.Model):705 """706 Workflow App execution log, excluding workflow debugging records.707 708 Attributes:709 710 - id (uuid) run ID711 - tenant_id (uuid) Workspace ID712 - app_id (uuid) App ID713 - workflow_id (uuid) Associated Workflow ID714 - workflow_run_id (uuid) Associated Workflow Run ID715 - created_from (string) Creation source716 717 `service-api` App Execution OpenAPI718 719 `web-app` WebApp720 721 `installed-app` Installed App722 723 - created_by_role (string) Creator role724 725 - `account` Console account726 727 - `end_user` End user728 729 - created_by (uuid) Creator ID, depends on the user table according to created_by_role730 - created_at (timestamp) Creation time731 """732 733 __tablename__ = "workflow_app_logs"734 __table_args__ = (735 db.PrimaryKeyConstraint("id", name="workflow_app_log_pkey"),736 db.Index("workflow_app_log_app_idx", "tenant_id", "app_id"),737 )738 739 id = db.Column(StringUUID, server_default=db.text("uuid_generate_v4()"))740 tenant_id = db.Column(StringUUID, nullable=False)741 app_id = db.Column(StringUUID, nullable=False)742 workflow_id = db.Column(StringUUID, nullable=False)743 workflow_run_id = db.Column(StringUUID, nullable=False)744 created_from = db.Column(db.String(255), nullable=False)745 created_by_role = db.Column(db.String(255), nullable=False)746 created_by = db.Column(StringUUID, nullable=False)747 created_at = db.Column(db.DateTime, nullable=False, server_default=db.text("CURRENT_TIMESTAMP(0)"))748 749 @property750 def workflow_run(self):751 return db.session.get(WorkflowRun, self.workflow_run_id)752 753 @property754 def created_by_account(self):755 created_by_role = CreatedByRole(self.created_by_role)756 return db.session.get(Account, self.created_by) if created_by_role == CreatedByRole.ACCOUNT else None757 758 @property759 def created_by_end_user(self):760 from models.model import EndUser761 762 created_by_role = CreatedByRole(self.created_by_role)763 return db.session.get(EndUser, self.created_by) if created_by_role == CreatedByRole.END_USER else None764 765 766class ConversationVariable(db.Model):767 __tablename__ = "workflow_conversation_variables"768 769 id: Mapped[str] = db.Column(StringUUID, primary_key=True)770 conversation_id: Mapped[str] = db.Column(StringUUID, nullable=False, primary_key=True)771 app_id: Mapped[str] = db.Column(StringUUID, nullable=False, index=True)772 data = db.Column(db.Text, nullable=False)773 created_at = db.Column(db.DateTime, nullable=False, index=True, server_default=db.text("CURRENT_TIMESTAMP(0)"))774 updated_at = db.Column(775 db.DateTime, nullable=False, server_default=func.current_timestamp(), onupdate=func.current_timestamp()776 )777 778 def __init__(self, *, id: str, app_id: str, conversation_id: str, data: str) -> None:779 self.id = id780 self.app_id = app_id781 self.conversation_id = conversation_id782 self.data = data783 784 @classmethod785 def from_variable(cls, *, app_id: str, conversation_id: str, variable: Variable) -> "ConversationVariable":786 obj = cls(787 id=variable.id,788 app_id=app_id,789 conversation_id=conversation_id,790 data=variable.model_dump_json(),791 )792 return obj793 794 def to_variable(self) -> Variable:795 mapping = json.loads(self.data)796 return variable_factory.build_variable_from_mapping(mapping)797