feat: add s3 file storage implementation (#10526)

* Beginning cherry-pick of changes

* more updates

* Add local and s3 storage services, modified base storage service, delegate from langflow to lfx

* more change

* remove all instances of with session

* test s3

* add tests and fix fe

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

* fix diamond inheritance and some flush / reset db patterns

* [autofix.ci] apply automated fixes

* Add methods for injectable session scopes

* user run_until_complete for some async calls

* [autofix.ci] apply automated fixes

* Fix csv and json to data components

* Fix some imports and add s3_comp and local-storage_serv testing

* Fix test paths

* more test fixes

* Import and test fixes

* fix remaining s3 compo tests

* Fix langchain compatibility versions

* Fix json/csv agent test mocks

* Fix base file reads with structured output path

* re-add lock

* Add safety around file streaming gen

* [autofix.ci] apply automated fixes

* comment

* improve error handling for file deletion

* abstract reads to storage utils

* [autofix.ci] apply automated fixes

* remove unnecessary deps

* fix lint issues

* fix byte stream generator call in download_file function

* Add import error handling for langchain dependencies in CSV and JSON agent components

* Add pypdf as a dependency in both uv.lock and pyproject.toml files

* ruff

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

* fix test patches

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

* [autofix.ci] apply automated fixes (attempt 3/3)

* Add noop database service and fix tests to utilize that

* comp index

* mock the imports more but not sure if that's the best soln

* ruff

* [autofix.ci] apply automated fixes

* uses commit in test since it yields

* [autofix.ci] apply automated fixes

* Fix prof pic test

* update prof pic size to fix test?

* ruff

* update lockfile

* update comp index

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

* use full path to altk test import

* update lockfile. again?

* Marks json test to allow a task leak, allowing since deprecated component

* [autofix.ci] apply automated fixes

* use lockfile from main to temporarily bypass issues with altk

* Update component build index

* [autofix.ci] apply automated fixes

* Fix conftest session scope usage?)

* Fix conftest yields

* lockfile

* [autofix.ci] apply automated fixes

* Fix another test session patterns

* Fix test type

* ruff

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes

* Update s3.py

* [autofix.ci] apply automated fixes

* Document the append param

* [autofix.ci] apply automated fixes

* Update test_s3_endpoints.py

* [autofix.ci] apply automated fixes

* add back dep import script

* mock fixes

* fix save file tests

* [autofix.ci] apply automated fixes

* Flush after delete in upload_user_file

* use current path in loop csv test

* ruff

* ruff

* Updates the sqlite pragma test to use sqlite3 directly

* ruff

* ensure pragma test db is in wal mode

* Adds back some relevant tests

* ruff

* Add more integration s3 tests

* ruff

* component index

---------

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Gabriel Luiz Freitas Almeida <[email protected]>
Co-authored-by: Eric Hare <[email protected]>

Author: 4 people

.secrets.baseline

@@ -883,7 +883,7 @@
         "filename": "src/backend/tests/unit/api/v2/test_files.py",
         "hashed_secret": "61fbb5a12cd7b1f1fe1624120089efc0cd299e43",
         "is_verified": false,
-        "line_number": 29,
+        "line_number": 40,
         "is_secret": false
       }
     ],
@@ -1412,5 +1412,5 @@
       }
     ]
   },
-  "generated_at": "2025-11-10T17:33:14Z"
+  "generated_at": "2025-11-19T18:36:04Z"
 }

docs/docs/Develop/memory.mdx

@@ -80,6 +80,8 @@ To fine-tune your database connection pool and timeout settings, you can set the
 
 * `LANGFLOW_DB_CONNECT_TIMEOUT`: The number of seconds to wait before giving up on a lock to be released or establishing a connection to the database. This may be separate from the `pool_timeout` in `LANGFLOW_DB_CONNECTION_SETTINGS`. Default: 30.
 
+* `LANGFLOW_MIGRATION_LOCK_NAMESPACE`: Optional namespace identifier for PostgreSQL advisory lock during migrations. If not provided, a hash of the database URL will be used. Useful when multiple Langflow instances share the same database and need coordinated migration locking
+
 * `LANGFLOW_DB_CONNECTION_SETTINGS`: A JSON dictionary containing the following database connection pool settings:
 
     - `pool_size`: The base number of connections to keep open in the connection pool. Default: 20.

pyproject.toml

@@ -75,7 +75,7 @@ dependencies = [
     "duckduckgo_search==7.2.1",
     "opensearch-py==2.8.0",
     "langchain-google-genai==2.0.6",
-    "langchain-cohere==0.3.3",
+    "langchain-cohere>=0.3.3,<1.0.0",
     "langchain-huggingface==0.3.1",
     "langchain-anthropic==0.3.14",
     "langchain-astradb>=0.6.1,<1.0.0",
@@ -85,14 +85,14 @@ dependencies = [
     "langchain-pinecone>=0.2.8,<1.0.0",
     "langchain-mistralai==0.2.3",
     "langchain-chroma>=0.2.6,<1.0.0",
-    "langchain-aws==0.2.33",
+    "langchain-aws>=0.2.33,<1.0.0",
     "langchain-unstructured==0.1.5",
     "langchain-milvus==0.1.7",
     "langchain-mongodb==0.7.0",
     "langchain-nvidia-ai-endpoints==0.3.8",
     "langchain-google-calendar-tools==0.0.1",
-    "langchain-google-community==2.0.3",
-    "langchain-elasticsearch==0.3.0",
+    "langchain-google-community>=2.0.3,<3.0.0",
+    "langchain-elasticsearch>=0.3.0,<1.0.0",
     "langchain-ollama==0.3.10",
     "langchain-sambanova==0.1.0",
     "langchain-community>=0.3.21,<1.0.0",
@@ -138,6 +138,7 @@ dependencies = [
     "cuga==0.1.10",
     "agent-lifecycle-toolkit~=0.4.1",
     "astrapy>=2.1.0,<3.0.0",
+    "aioboto3>=15.2.0,<16.0.0"
 ]
 
 

src/backend/base/langflow/__main__.py

@@ -891,9 +891,7 @@ async def aapi_key():
                 await delete_api_key(session, api_key.id)
 
             api_key_create = ApiKeyCreate(name="CLI")
-            unmasked_api_key = await create_api_key(session, api_key_create, user_id=superuser.id)
-            await session.commit()
-            return unmasked_api_key
+            return await create_api_key(session, api_key_create, user_id=superuser.id)
 
     unmasked_api_key = asyncio.run(aapi_key())
     # Create a banner to display the API key and tell the user it won't be shown again

src/backend/base/langflow/alembic/env.py

@@ -1,15 +1,21 @@
 # noqa: INP001
 import asyncio
+import hashlib
+import os
 from logging.config import fileConfig
 from typing import Any
 
+
 from alembic import context
 from sqlalchemy import pool, text
 from sqlalchemy.event import listen
 from sqlalchemy.ext.asyncio import async_engine_from_config
 
+from lfx.log.logger import logger
+
 from langflow.services.database.service import SQLModel
 
+
 # this is the Alembic Config object, which provides
 # access to the values within the .ini file in use.
 config = context.config
@@ -96,25 +102,33 @@ def _do_run_migrations(connection):
         configure_kwargs["prepare_threshold"] = None
 
     context.configure(**configure_kwargs)
-
     with context.begin_transaction():
         if connection.dialect.name == "postgresql":
-            connection.execute(text("SET LOCAL lock_timeout = '60s';"))
-            connection.execute(text("SELECT pg_advisory_xact_lock(112233);"))
+            # Use namespace from environment variable if provided, otherwise use default static key
+            namespace = os.getenv("LANGFLOW_MIGRATION_LOCK_NAMESPACE")
+            if namespace:
+                lock_key = int(hashlib.sha256(namespace.encode()).hexdigest()[:16], 16) % (2**63 - 1)
+                logger.info(f"Using migration lock namespace: {namespace}, lock_key: {lock_key}")
+            else:
+                lock_key = 11223344
+                logger.info(f"Using default migration lock_key: {lock_key}")
+
+            connection.execute(text("SET LOCAL lock_timeout = '180s';"))
+            connection.execute(text(f"SELECT pg_advisory_xact_lock({lock_key});"))
         context.run_migrations()
 
-
 async def _run_async_migrations() -> None:
-    # Get database URL to determine dialect
-    url = config.get_main_option("sqlalchemy.url")
-    connect_args: dict[str, Any] = {}
+    # Disable prepared statements for PostgreSQL (required for PgBouncer compatibility)
+    # SQLite doesn't support this parameter, so only add it for PostgreSQL
+    config_section = config.get_section(config.config_ini_section, {})
+    db_url = config_section.get("sqlalchemy.url", "")
 
-    # Only add prepare_threshold for PostgreSQL
-    if url and "postgresql" in url:
+    connect_args: dict[str, Any] = {}
+    if db_url and "postgresql" in db_url:
         connect_args["prepare_threshold"] = None
 
     connectable = async_engine_from_config(
-        config.get_section(config.config_ini_section, {}),
+        config_section,
         prefix="sqlalchemy.",
         poolclass=pool.NullPool,
         connect_args=connect_args,

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

@@ -10,6 +10,7 @@
 from fastapi_pagination import Params
 from lfx.graph.graph.base import Graph
 from lfx.log.logger import logger
+from lfx.services.deps import injectable_session_scope, injectable_session_scope_readonly, session_scope
 from sqlalchemy import delete
 from sqlmodel.ext.asyncio.session import AsyncSession
 
@@ -19,7 +20,6 @@
 from langflow.services.database.models.transactions.model import TransactionTable
 from langflow.services.database.models.user.model import User
 from langflow.services.database.models.vertex_builds.model import VertexBuildTable
-from langflow.services.deps import get_session, session_scope
 from langflow.services.store.utils import get_lf_version_from_pypi
 from langflow.utils.constants import LANGFLOW_GLOBAL_VAR_HEADER_PREFIX
 
@@ -35,7 +35,10 @@
 
 CurrentActiveUser = Annotated[User, Depends(get_current_active_user)]
 CurrentActiveMCPUser = Annotated[User, Depends(get_current_active_user_mcp)]
-DbSession = Annotated[AsyncSession, Depends(get_session)]
+# DbSession with auto-commit for write operations
+DbSession = Annotated[AsyncSession, Depends(injectable_session_scope)]
+# DbSessionReadOnly for read-only operations (no auto-commit, reduces lock contention)
+DbSessionReadOnly = Annotated[AsyncSession, Depends(injectable_session_scope_readonly)]
 
 
 class EventDeliveryType(str, Enum):

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

@@ -289,6 +289,7 @@ async def build_vertex(
         if isinstance(cache, CacheMiss):
             # If there's no cache
             await logger.awarning(f"No cache found for {flow_id_str}. Building graph starting at {vertex_id}")
+
             async with session_scope() as session:
                 graph = await build_graph_from_db(
                     flow_id=flow_id,

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

@@ -173,10 +173,9 @@ async def list_profile_pictures(
         people_path = config_path / "profile_pictures" / "People"
         space_path = config_path / "profile_pictures" / "Space"
 
-        # List files directly from local filesystem
+        # List files directly from local filesystem - bundled with the container
         people = [f.name for f in people_path.iterdir() if f.is_file()] if people_path.exists() else []
         space = [f.name for f in space_path.iterdir() if f.is_file()] if space_path.exists() else []
-
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e)) from e
 

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

@@ -160,11 +160,12 @@ async def create_flow(
 ):
     try:
         db_flow = await _new_flow(session=session, flow=flow, user_id=current_user.id)
-        await session.commit()
+        await session.flush()
         await session.refresh(db_flow)
-
         await _save_flow_to_fs(db_flow)
 
+        # Convert to FlowRead while session is still active to avoid detached instance errors
+        flow_read = FlowRead.model_validate(db_flow, from_attributes=True)
     except Exception as e:
         if "UNIQUE constraint failed" in str(e):
             # Get the name of the column that failed
@@ -180,7 +181,7 @@ async def create_flow(
         if isinstance(e, HTTPException):
             raise
         raise HTTPException(status_code=500, detail=str(e)) from e
-    return db_flow
+    return flow_read
 
 
 @router.get("/", response_model=list[FlowRead] | Page[FlowRead] | list[FlowHeader], status_code=200)
@@ -258,8 +259,9 @@ async def read_flows(
                 flow_headers = [FlowHeader.model_validate(flow, from_attributes=True) for flow in flows]
                 return compress_response(flow_headers)
 
-            # Compress the full flows response
-            return compress_response(flows)
+            # Convert to FlowRead while session is still active to avoid detached instance errors
+            flow_reads = [FlowRead.model_validate(flow, from_attributes=True) for flow in flows]
+            return compress_response(flow_reads)
 
         stmt = stmt.where(Flow.folder_id == folder_id)
 
@@ -295,7 +297,8 @@ async def read_flow(
 ):
     """Read a flow."""
     if user_flow := await _read_flow(session, flow_id, current_user.id):
-        return user_flow
+        # Convert to FlowRead while session is still active to avoid detached instance errors
+        return FlowRead.model_validate(user_flow, from_attributes=True)
     raise HTTPException(status_code=404, detail="Flow not found")
 
 
@@ -358,11 +361,13 @@ async def update_flow(
                 db_flow.folder_id = default_folder.id
 
         session.add(db_flow)
-        await session.commit()
+        await session.flush()
         await session.refresh(db_flow)
-
         await _save_flow_to_fs(db_flow)
 
+        # Convert to FlowRead while session is still active to avoid detached instance errors
+        flow_read = FlowRead.model_validate(db_flow, from_attributes=True)
+
     except Exception as e:
         if "UNIQUE constraint failed" in str(e):
             # Get the name of the column that failed
@@ -379,7 +384,7 @@ async def update_flow(
             raise HTTPException(status_code=e.status_code, detail=str(e)) from e
         raise HTTPException(status_code=500, detail=str(e)) from e
 
-    return db_flow
+    return flow_read
 
 
 @router.delete("/{flow_id}", status_code=200)
@@ -398,7 +403,6 @@ async def delete_flow(
     if not flow:
         raise HTTPException(status_code=404, detail="Flow not found")
     await cascade_delete_flow(session, flow.id)
-    await session.commit()
     return {"message": "Flow deleted successfully"}
 
 
@@ -416,10 +420,12 @@ async def create_flows(
         db_flow = Flow.model_validate(flow, from_attributes=True)
         session.add(db_flow)
         db_flows.append(db_flow)
-    await session.commit()
+
+    await session.flush()
     for db_flow in db_flows:
         await session.refresh(db_flow)
-    return db_flows
+
+    return [FlowRead.model_validate(db_flow, from_attributes=True) for db_flow in db_flows]
 
 
 @router.post("/upload/", response_model=list[FlowRead], status_code=201)
@@ -444,10 +450,13 @@ async def upload_file(
         response_list.append(response)
 
     try:
-        await session.commit()
+        await session.flush()
         for db_flow in response_list:
             await session.refresh(db_flow)
             await _save_flow_to_fs(db_flow)
+
+        # Convert to FlowRead while session is still active to avoid detached instance errors
+        flow_reads = [FlowRead.model_validate(db_flow, from_attributes=True) for db_flow in response_list]
     except Exception as e:
         if "UNIQUE constraint failed" in str(e):
             # Get the name of the column that failed
@@ -464,7 +473,7 @@ async def upload_file(
             raise
         raise HTTPException(status_code=500, detail=str(e)) from e
 
-    return response_list
+    return flow_reads
 
 
 @router.delete("/")
@@ -491,7 +500,7 @@ async def delete_multiple_flows(
         for flow in flows_to_delete:
             await cascade_delete_flow(db, flow.id)
 
-        await db.commit()
+        await db.flush()
         return {"deleted": len(flows_to_delete)}
     except Exception as exc:
         raise HTTPException(status_code=500, detail=str(exc)) from exc

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

@@ -440,6 +440,8 @@ async def update_project_mcp_settings(
                     session.add(flow)
                     updated_flows.append(flow)
 
+            await session.flush()
+
             response: dict[str, Any] = {
                 "message": f"Updated MCP settings for {len(updated_flows)} flows and project auth settings"
             }
@@ -1201,8 +1203,6 @@ async def init_mcp_servers():
 
             # Auto-configure starter projects with MCP server settings if enabled
             await auto_configure_starter_projects_mcp(session)
-            # Commit any auth settings updates
-            await session.commit()
 
     except Exception as e:  # noqa: BLE001
         msg = f"Failed to initialize MCP servers: {e}"