feat: add s3 implementation for storage layer

docs/docs/Configuration/environment-variables.mdx

@@ -88,6 +88,7 @@ If it detects a supported environment variable, then it automatically adopts the
     LANGFLOW_CONFIG_DIR=/path/to/config/
     LANGFLOW_DATABASE_URL=postgresql://user:password@localhost:5432/langflow
     LANGFLOW_DEV=False
+    LANGFLOW_MIGRATION_LOCK_NAMESPACE=my-namespace
     LANGFLOW_FALLBACK_TO_ENV_VAR=False
     LANGFLOW_HEALTH_CHECK_MAX_RETRIES=5
     LANGFLOW_HOST=localhost

docs/docs/Develop/memory.mdx

@@ -61,6 +61,7 @@ The following settings allow you to fine-tune your database connection pool and
 ```
 LANGFLOW_DB_CONNECTION_SETTINGS='{"pool_size": 20, "max_overflow": 30, "pool_timeout": 30, "pool_pre_ping": true, "pool_recycle": 1800, "echo": false}'
 LANGFLOW_DB_CONNECT_TIMEOUT=20
+LANGFLOW_MIGRATION_LOCK_NAMESPACE=my-namespace
 ```
 
 - `pool_size`: Maximum number of database connections to keep in the pool (default: 20)
@@ -70,6 +71,7 @@ LANGFLOW_DB_CONNECT_TIMEOUT=20
 - `pool_recycle`: Number of seconds after which a connection is automatically recycled (default: 1800, or 30 minutes)
 - `echo`: If true, SQL queries are logged for debugging purposes (default: false)
 - `LANGFLOW_DB_CONNECT_TIMEOUT`: Maximum number of seconds to wait when establishing a new database connection (default: 20)
+- `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
 
 ## Configure cache memory
 

pyproject.toml

@@ -85,7 +85,7 @@ 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",
@@ -133,6 +133,7 @@ dependencies = [
     "easyocr>=1.7.2; sys_platform != 'darwin' or platform_machine != 'x86_64'",
     "opencv-python>=4.11; sys_platform != 'darwin' or platform_machine != 'x86_64'",
     "pytest-codspeed==4.0.0", # Pinned to avoid python3.13 + intel mac compatibility issues
+    "aioboto3>=15.2.0,<16.0.0",
 ]
 
 [dependency-groups]

src/backend/base/langflow/__main__.py

@@ -877,8 +877,7 @@
 
             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
 
     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,5 +1,8 @@
 # noqa: INP001
 import asyncio
+import hashlib
+import logging
+import os
 from logging.config import fileConfig
 
 from alembic import context
@@ -9,6 +12,8 @@
 
 from langflow.services.database.service import SQLModel
 
+logger = logging.getLogger(__name__)
+
 # this is the Alembic Config object, which provides
 # access to the values within the .ini file in use.
 config = context.config
@@ -85,16 +90,34 @@
 
     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:
+    # 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", "")
+
+    connect_args = {}
+    if "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,
     )
 
     if connectable.dialect.name == "sqlite":

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

@@ -19,7 +19,7 @@
 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.deps import get_session, get_session_with_commit
 from langflow.services.store.utils import get_lf_version_from_pypi
 
 if TYPE_CHECKING:
@@ -34,7 +34,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(get_session_with_commit)]
+# DbSessionReadOnly for read-only operations (no auto-commit, reduces lock contention)
+DbSessionReadOnly = Annotated[AsyncSession, Depends(get_session)]
 
 
 class EventDeliveryType(str, Enum):
@@ -150,7 +153,7 @@
 
 
 async def _get_flow_name(flow_id: uuid.UUID) -> str:
     async with session_scope() as session:
         flow = await session.get(Flow, flow_id)
         if flow is None:
             msg = f"Flow {flow_id} not found"
@@ -351,7 +354,7 @@
         raise HTTPException(status_code=400, detail="No client_id cookie found")
 
     # Check if the flow is public
     async with session_scope() as session:
         from sqlmodel import select
 
         from langflow.services.database.models.flow.model import AccessTypeEnum, Flow

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

@@ -43,7 +43,6 @@
 from langflow.services.deps import (
     get_chat_service,
     get_queue_service,
-    get_session,
     get_telemetry_service,
     session_scope,
 )
@@ -286,11 +285,14 @@ 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}")
-            graph = await build_graph_from_db(
-                flow_id=flow_id,
-                session=await anext(get_session()),
-                chat_service=chat_service,
-            )
+            from langflow.services.deps import session_scope
+
+            async with session_scope() as session:
+                graph = await build_graph_from_db(
+                    flow_id=flow_id,
+                    session=session,
+                    chat_service=chat_service,
+                )
         else:
             graph = cache.get("result")
             await graph.initialize_run()

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

@@ -130,14 +130,20 @@
     folder_name: str,
     file_name: str,
 ):
+    """Download a profile picture.
+    Profile pictures are system files bundled with the package, served from the installation directory.
+    """
     try:
-        storage_service = get_storage_service()
+        # Profile pictures are in the package installation directory
+        package_dir = Path(__file__).parent.parent.parent / "initial_setup" / "profile_pictures"
+        file_path = package_dir / folder_name / file_name
+
+        if not file_path.exists():
+            raise HTTPException(status_code=404, detail="Profile picture not found")
+
         extension = file_name.split(".")[-1]
-        config_dir = storage_service.settings_service.settings.config_dir
-        config_path = Path(config_dir)  # type: ignore[arg-type]
-        folder_path = config_path / "profile_pictures" / folder_name
         content_type = build_content_type_from_extension(extension)
-        file_content = await storage_service.get_file(flow_id=folder_path, file_name=file_name)  # type: ignore[arg-type]
+        file_content = file_path.read_bytes()
         return StreamingResponse(BytesIO(file_content), media_type=content_type)
 
     except Exception as e:
@@ -146,16 +152,19 @@
 
 @router.get("/profile_pictures/list")
 async def list_profile_pictures():
+    """List available profile pictures.
+    Profile pictures are system files bundled with the package, served from the installation directory.
+    """
     try:
-        storage_service = get_storage_service()
-        config_dir = storage_service.settings_service.settings.config_dir
-        config_path = Path(config_dir)  # type: ignore[arg-type]
+        # Profile pictures are in the package installation directory
+        package_dir = Path(__file__).parent.parent.parent / "initial_setup" / "profile_pictures"
 
-        people_path = config_path / "profile_pictures/People"
-        space_path = config_path / "profile_pictures/Space"
+        people_path = package_dir / "People"
+        space_path = package_dir / "Space"
 
-        people = await storage_service.list_files(flow_id=people_path)  # type: ignore[arg-type]
-        space = await storage_service.list_files(flow_id=space_path)  # type: ignore[arg-type]
+        # List files from package directory - these are 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,14 @@
 ):
     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 +183,7 @@
         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)
@@ -358,7 +361,7 @@
                 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)
@@ -398,7 +401,6 @@
     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 +418,12 @@
         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
+    # 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 db_flows]
+    return flow_reads
 
 
 @router.post("/upload/", response_model=list[FlowRead], status_code=201)
@@ -444,7 +448,7 @@
         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)

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

@@ -416,8 +416,6 @@ async def update_project_mcp_settings(
                     session.add(flow)
                     updated_flows.append(flow)
 
-            await session.commit()
-
             response: dict[str, Any] = {
                 "message": f"Updated MCP settings for {len(updated_flows)} flows and project auth settings"
             }
@@ -1176,9 +1174,6 @@ async def init_mcp_servers():
                     await logger.aexception(msg)
                     # Continue to next project even if this one fails
 
-            # Commit any auth settings updates
-            await session.commit()
-
     except Exception as e:  # noqa: BLE001
         msg = f"Failed to initialize MCP servers: {e}"
         await logger.aexception(msg)

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

@@ -35,7 +35,6 @@ async def get_vertex_builds(flow_id: Annotated[UUID, Query()], session: DbSessio
 async def delete_vertex_builds(flow_id: Annotated[UUID, Query()], session: DbSession) -> None:
     try:
         await delete_vertex_builds_by_flow_id(session, flow_id)
-        await session.commit()
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e)) from e
 
@@ -93,7 +92,6 @@ async def get_messages(
 async def delete_messages(message_ids: list[UUID], session: DbSession) -> None:
     try:
         await session.exec(delete(MessageTable).where(MessageTable.id.in_(message_ids)))  # type: ignore[attr-defined]
-        await session.commit()
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e)) from e
 
@@ -118,7 +116,7 @@ async def update_message(
             message_dict["edit"] = True
         db_message.sqlmodel_update(message_dict)
         session.add(db_message)
-        await session.commit()
+        await session.flush()
         await session.refresh(db_message)
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e)) from e
@@ -151,7 +149,7 @@ async def update_session_id(
 
         session.add_all(messages)
 
-        await session.commit()
+        await session.flush()
         message_responses = []
         for message in messages:
             await session.refresh(message)
@@ -173,7 +171,6 @@ async def delete_messages_session(
             .where(col(MessageTable.session_id) == session_id)
             .execution_options(synchronize_session="fetch")
         )
-        await session.commit()
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e)) from e