feat: add option to skip message persistence in Component

pyproject.toml

@@ -296,7 +296,7 @@ ignore = [
     "TD002", # Missing author in TODO
     "TD003", # Missing issue link in TODO
     "TRY301", # A bit too harsh (Abstract `raise` to an inner function)
-
+    "PLC0415",
     # Rules that are TODOs
     "ANN",
 ]

src/backend/base/langflow/api/build.py

@@ -11,6 +11,7 @@
 
 from langflow.api.disconnect import DisconnectHandlerStreamingResponse
 from langflow.api.utils import (
+    ComponentConfig,
     CurrentActiveUser,
     EventDeliveryType,
     build_graph_from_data,
@@ -51,6 +52,7 @@ async def start_flow_build(
     current_user: CurrentActiveUser,
     queue_service: JobQueueService,
     flow_name: str | None = None,
+    component_config: ComponentConfig | None = None,
 ) -> str:
     """Start the flow build process by setting up the queue and starting the build task.
 
@@ -72,6 +74,7 @@ async def start_flow_build(
             log_builds=log_builds,
             current_user=current_user,
             flow_name=flow_name,
+            component_config=component_config,
         )
         queue_service.start_job(job_id, task_coro)
     except Exception as e:
@@ -191,6 +194,7 @@ async def generate_flow_events(
     log_builds: bool,
     current_user: CurrentActiveUser,
     flow_name: str | None = None,
+    component_config: ComponentConfig | None = None,
 ) -> None:
     """Generate events for flow building process.
 
@@ -263,6 +267,7 @@ async def create_graph(fresh_session, flow_id_str: str, flow_name: str | None) -
                 chat_service=chat_service,
                 user_id=str(current_user.id),
                 session_id=effective_session_id,
+                component_config=component_config,
             )
 
         if not flow_name:

src/backend/base/langflow/api/utils.py

@@ -9,6 +9,7 @@
 from fastapi import Depends, HTTPException, Query
 from fastapi_pagination import Params
 from loguru import logger
+from pydantic import BaseModel
 from sqlalchemy import delete
 from sqlmodel.ext.asyncio.session import AsyncSession
 
@@ -43,6 +44,10 @@ class EventDeliveryType(str, Enum):
     POLLING = "polling"
 
 
+class ComponentConfig(BaseModel):
+    persist_messages: bool = True
+
+
 def has_api_terms(word: str):
     return "api" in word and ("key" in word or ("token" in word and "tokens" not in word))
 
@@ -168,7 +173,19 @@ async def build_graph_from_data(flow_id: uuid.UUID | str, payload: dict, **kwarg
     str_flow_id = str(flow_id)
     session_id = kwargs.get("session_id") or str_flow_id
 
-    graph = Graph.from_payload(payload, str_flow_id, flow_name, kwargs.get("user_id"))
+    component_config = kwargs.get("component_config")
+    if component_config is None:
+        component_config = {}
+    if isinstance(component_config, ComponentConfig):
+        component_config = component_config.model_dump()
+
+    graph = Graph.from_payload(
+        payload,
+        str_flow_id,
+        flow_name,
+        user_id=kwargs.get("user_id"),
+        component_config=component_config,
+    )
     for vertex_id in graph.has_session_id_vertices:
         vertex = graph.get_vertex(vertex_id)
         if vertex is None:

src/backend/base/langflow/api/v1/chat.py

@@ -25,6 +25,7 @@
 )
 from langflow.api.limited_background_tasks import LimitVertexBuildBackgroundTasks
 from langflow.api.utils import (
+    ComponentConfig,
     CurrentActiveUser,
     DbSession,
     EventDeliveryType,
@@ -154,6 +155,7 @@ async def build_flow(
     queue_service: Annotated[JobQueueService, Depends(get_queue_service)],
     flow_name: str | None = None,
     event_delivery: EventDeliveryType = EventDeliveryType.POLLING,
+    component_config: Annotated[ComponentConfig | None, Body(embed=True)] = None,
 ):
     """Build and process a flow, returning a job ID for event polling.
 
@@ -173,6 +175,7 @@ async def build_flow(
         queue_service: Queue service for job management
         flow_name: Optional name for the flow
         event_delivery: Optional event delivery type - default is streaming
+        component_config: Optional component configuration
 
     Returns:
         Dict with job_id that can be used to poll for build status
@@ -195,6 +198,7 @@ async def build_flow(
         current_user=current_user,
         queue_service=queue_service,
         flow_name=flow_name,
+        component_config=component_config,
     )
 
     # This is required to support FE tests - we need to be able to set the event delivery to direct
@@ -573,6 +577,7 @@ async def build_public_tmp(
     request: Request,
     queue_service: Annotated[JobQueueService, Depends(get_queue_service)],
     event_delivery: EventDeliveryType = EventDeliveryType.POLLING,
+    component_config: Annotated[ComponentConfig | None, Body(embed=True)] = None,
 ):
     """Build a public flow without requiring authentication.
 
@@ -601,6 +606,7 @@ async def build_public_tmp(
         request: FastAPI request object (needed for cookie access)
         queue_service: Queue service for job management
         event_delivery: Optional event delivery type - default is streaming
+        component_config: Optional component configuration
 
     Returns:
         Dict with job_id that can be used to poll for build status
@@ -623,6 +629,7 @@ async def build_public_tmp(
             current_user=owner_user,
             queue_service=queue_service,
             flow_name=flow_name or f"{client_id}_{flow_id}",
+            component_config=component_config,
         )
     except Exception as exc:
         logger.exception("Error building public flow")

src/backend/base/langflow/api/v1/endpoints.py

@@ -682,7 +682,9 @@ async def custom_component_update(
 ):
     """Update an existing custom component with new code and configuration.
 
-    Processes the provided code and template updates, applies parameter changes (including those loaded from the database), updates the component's build configuration, and validates outputs. Returns the updated component node as a JSON-serializable dictionary.
+    Processes the provided code and template updates, applies parameter changes (including those loaded from the
+    database), updates the component's build configuration, and validates outputs. Returns the updated component node
+    as a JSON-serializable dictionary.
 
     Raises:
         HTTPException: If an error occurs during component building or updating.

src/backend/base/langflow/custom/custom_component/component.py

@@ -35,6 +35,7 @@
 from langflow.schema.data import Data
 from langflow.schema.message import ErrorMessage, Message
 from langflow.schema.properties import Source
+from langflow.services.deps import get_db_service
 from langflow.services.tracing.schema import Log
 from langflow.template.field.base import UNDEFINED, Input, Output
 from langflow.template.frontend_node.custom_components import ComponentFrontendNode
@@ -162,6 +163,7 @@ def __init__(self, **kwargs) -> None:
         # Final setup
         self._set_output_types(list(self._outputs_map.values()))
         self.set_class_code()
+        self._database_available: bool | None = None
 
     def _build_source(self, id_: str | None, display_name: str | None, source: str | None) -> Source:
         source_dict = {}
@@ -1460,6 +1462,16 @@ def is_connected_to_chat_output(self) -> bool:
 
         return has_chat_output(self.graph.get_vertex_neighbors(self._vertex))
 
+    def _is_database_available(self) -> bool:
+        """Check if the database is available."""
+        if (
+            hasattr(self, "graph")
+            and hasattr(self.graph, "component_config")
+            and not self.graph.component_config.get("persist_messages", True)
+        ):
+            return False
+        return get_db_service().database_available
+
     def _should_skip_message(self, message: Message) -> bool:
         """Check if the message should be skipped based on vertex configuration and message type."""
         return (
@@ -1479,25 +1491,32 @@ async def send_message(self, message: Message, id_: str | None = None):
             message.session_id = session_id
         if hasattr(message, "flow_id") and isinstance(message.flow_id, str):
             message.flow_id = UUID(message.flow_id)
-        stored_message = await self._store_message(message)
+        if self._is_database_available():
+            stored_message = await self._store_message(message)
+            self._stored_message_id = stored_message.id
+        else:
+            stored_message = message
 
-        self._stored_message_id = stored_message.id
         try:
             complete_message = ""
             if (
-                self._should_stream_message(stored_message, message)
+                self._should_stream_message(message)
                 and message is not None
                 and isinstance(message.text, AsyncIterator | Iterator)
             ):
                 complete_message = await self._stream_message(message.text, stored_message)
                 stored_message.text = complete_message
-                stored_message = await self._update_stored_message(stored_message)
+                if self._is_database_available():
+                    stored_message = await self._update_stored_message(stored_message)
+                else:
+                    stored_message = message
             else:
                 # Only send message event for non-streaming messages
                 await self._send_message_event(stored_message, id_=id_)
         except Exception:
             # remove the message from the database
-            await delete_message(stored_message.id)
+            if self._is_database_available():
+                await delete_message(stored_message.id)
             raise
         self.status = stored_message
         return stored_message
@@ -1539,12 +1558,9 @@ def _send_event():
 
             await asyncio.to_thread(_send_event)
 
-    def _should_stream_message(self, stored_message: Message, original_message: Message) -> bool:
+    def _should_stream_message(self, original_message: Message) -> bool:
         return bool(
-            hasattr(self, "_event_manager")
-            and self._event_manager
-            and stored_message.id
-            and not isinstance(original_message.text, str)
+            hasattr(self, "_event_manager") and self._event_manager and not isinstance(original_message.text, str)
         )
 
     async def _update_stored_message(self, message: Message) -> Message:

src/backend/base/langflow/graph/graph/base.py

@@ -68,6 +68,7 @@ def __init__(
         user_id: str | None = None,
         log_config: LogConfig | None = None,
         context: dict[str, Any] | None = None,
+        component_config: dict[str, Any] | None = None,
     ) -> None:
         """Initializes a new Graph instance.
 
@@ -125,6 +126,7 @@ def __init__(
         self._call_order: list[str] = []
         self._snapshots: list[dict[str, Any]] = []
         self._end_trace_tasks: set[asyncio.Task] = set()
+        self.component_config: dict[str, Any] = component_config or {}
 
         if context and not isinstance(context, dict):
             msg = "Context must be a dictionary"
@@ -1049,6 +1051,7 @@ def from_payload(
         flow_id: str | None = None,
         flow_name: str | None = None,
         user_id: str | None = None,
+        component_config: dict[str, Any] | None = None,
     ) -> Graph:
         """Creates a graph from a payload.
 
@@ -1057,6 +1060,7 @@ def from_payload(
             flow_id: The ID of the flow.
             flow_name: The flow name.
             user_id: The user ID.
+            component_config: Component configuration dictionary.
 
         Returns:
             Graph: The created graph.
@@ -1066,7 +1070,7 @@ def from_payload(
         try:
             vertices = payload["nodes"]
             edges = payload["edges"]
-            graph = cls(flow_id=flow_id, flow_name=flow_name, user_id=user_id)
+            graph = cls(flow_id=flow_id, flow_name=flow_name, user_id=user_id, component_config=component_config)
             graph.add_nodes_and_edges(vertices, edges)
         except KeyError as exc:
             logger.exception(exc)

src/backend/base/langflow/initial_setup/starter_projects/Vector Store RAG.json

@@ -4786,4 +4786,4 @@
     "rag",
     "q-a"
   ]
-}
+}

src/backend/base/langflow/interface/initialize/loading.py

@@ -44,6 +44,7 @@ def instantiate_class(
         _vertex=vertex,
         _tracing_service=get_tracing_service(),
         _id=vertex.id,
+        **vertex.graph.component_config,
     )
     if hasattr(custom_component, "set_event_manager"):
         custom_component.set_event_manager(event_manager)

src/backend/base/langflow/services/database/service.py

@@ -39,6 +39,7 @@ class DatabaseService(Service):
     name = "database_service"
 
     def __init__(self, settings_service: SettingsService):
+        self.engine: AsyncEngine | None = None
         self._logged_pragma = False
         self.settings_service = settings_service
         if settings_service.settings.database_url is None:
@@ -56,6 +57,7 @@ def __init__(self, settings_service: SettingsService):
         # register the event listener for sqlite as part of this class.
         # Using decorator will make the method not able to use self
         event.listen(Engine, "connect", self.on_connection)
+
         if self.settings_service.settings.database_connection_retry:
             self.engine = self._create_engine_with_retry()
         else:
@@ -68,6 +70,23 @@ def __init__(self, settings_service: SettingsService):
         else:
             self.alembic_log_path = Path(langflow_dir) / alembic_log_file
 
+        self._database_available = False
+
+    # Check if there's a message table in the database
+    async def _is_database_available(self) -> bool:
+        async with self.with_session() as session, session.bind.connect() as conn:
+            # Use run_sync to inspect the connection
+            def check_tables(conn):
+                inspector = inspect(conn)
+                return "message" in inspector.get_table_names()
+
+            self._database_available = await conn.run_sync(check_tables)
+        return self._database_available
+
+    @property
+    def database_available(self) -> bool:
+        return self._database_available
+
     async def initialize_alembic_log_file(self):
         # Ensure the directory and file for the alembic log file exists
         await anyio.Path(self.alembic_log_path.parent).mkdir(parents=True, exist_ok=True)
@@ -369,6 +388,7 @@ async def run_migrations(self, *, fix=False) -> None:
                 logger.debug("Alembic not initialized")
                 should_initialize_alembic = True
         await asyncio.to_thread(self._run_migrations, should_initialize_alembic, fix)
+        self._database_available = await self._is_database_available()
 
     @staticmethod
     def try_downgrade_upgrade_until_success(alembic_cfg, retries=5) -> None:
@@ -476,4 +496,6 @@ async def teardown(self) -> None:
                 await teardown_superuser(settings_service, session)
         except Exception:  # noqa: BLE001
             logger.exception("Error tearing down database")
-        await self.engine.dispose()
+        if self.engine is not None:
+            await self.engine.dispose()
+            self.engine = None

src/backend/tests/conftest.py

@@ -76,6 +76,10 @@ def blockbuster(request):
                 .can_block_in("rich/traceback.py", "_render_stack")
                 .can_block_in("langchain_core/_api/internal.py", "is_caller_internal")
                 .can_block_in("langchain_core/runnables/utils.py", "get_function_nonlocals")
+                .can_block_in("alembic/versions", "_load_revisions")
+                .can_block_in("dotenv/main.py", "find_dotenv")
+                .can_block_in("alembic/script/base.py", "_load_revisions")
+                .can_block_in("alembic/env.py", "_do_run_migrations")
             )
 
             for func in ["os.stat", "os.path.abspath", "os.scandir", "os.listdir"]: