perf(api): compress flow payloads on the wire

src/backend/base/langflow/alembic/versions/d3b7c1e05f84_compress_flow_version_data.py

@@ -0,0 +1,107 @@
+"""Store flow version snapshots gzipped.
+
+Phase: MIGRATE
+Revision ID: d3b7c1e05f84
+Revises: c9f2e5a7b1d4
+Create Date: 2026-09-02
+"""
+
+from __future__ import annotations
+
+import gzip
+import json
+from typing import TYPE_CHECKING
+
+import sqlalchemy as sa
+from alembic import op
+from langflow.utils import migration
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
+revision: str = "d3b7c1e05f84"  # pragma: allowlist secret
+down_revision: str | None = "c9f2e5a7b1d4"  # pragma: allowlist secret
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+BATCH_SIZE = 200
+COMPRESS_LEVEL = 6
+
+_flow_version = sa.table(
+    "flow_version",
+    sa.column("id"),
+    sa.column("data", sa.JSON()),
+    sa.column("data_gz", sa.LargeBinary()),
+)
+
+
+def _rows(conn, source: sa.Column):
+    offset = 0
+    while True:
+        stmt = (
+            sa.select(_flow_version.c.id, source)
+            .where(source.is_not(None))
+            .order_by(_flow_version.c.id)
+            .limit(BATCH_SIZE)
+            .offset(offset)
+        )
+        batch = conn.execute(stmt).fetchall()
+        if not batch:
+            return
+        yield batch
+        offset += BATCH_SIZE
+
+
+def _set_external_storage(conn) -> None:
+    if conn.dialect.name == "postgresql":
+        op.execute("ALTER TABLE flow_version ALTER COLUMN data_gz SET STORAGE EXTERNAL")
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    if not migration.table_exists("flow_version", conn):
+        return
+
+    if not migration.column_exists("flow_version", "data_gz", conn):
+        op.add_column("flow_version", sa.Column("data_gz", sa.LargeBinary(), nullable=True))
+    _set_external_storage(conn)
+
+    if migration.column_exists("flow_version", "data", conn):
+        migrated = 0
+        for batch in _rows(conn, _flow_version.c.data):
+            for row in batch:
+                payload = gzip.compress(
+                    json.dumps(row.data, separators=(",", ":"), ensure_ascii=False).encode("utf-8"),
+                    COMPRESS_LEVEL,
+                )
+                result = conn.execute(
+                    _flow_version.update().where(_flow_version.c.id == row.id).values(data_gz=payload)
+                )
+                migrated += result.rowcount
+        pending = conn.execute(
+            sa.select(sa.func.count())
+            .select_from(_flow_version)
+            .where(_flow_version.c.data.is_not(None), _flow_version.c.data_gz.is_(None))
+        ).scalar_one()
+        if pending:
+            msg = f"{pending} flow_version rows still hold uncompressed data after backfilling {migrated}"
+            raise RuntimeError(msg)
+        with op.batch_alter_table("flow_version") as batch_op:
+            batch_op.drop_column("data")
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    if not migration.table_exists("flow_version", conn):
+        return
+
+    if not migration.column_exists("flow_version", "data", conn):
+        op.add_column("flow_version", sa.Column("data", sa.JSON(), nullable=True))
+
+    if migration.column_exists("flow_version", "data_gz", conn):
+        for batch in _rows(conn, _flow_version.c.data_gz):
+            for row in batch:
+                payload = json.loads(gzip.decompress(row.data_gz))
+                conn.execute(_flow_version.update().where(_flow_version.c.id == row.id).values(data=payload))
+        with op.batch_alter_table("flow_version") as batch_op:
+            batch_op.drop_column("data_gz")

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

@@ -116,7 +116,6 @@
 )
 from langflow.services.event_manager import create_webhook_event_manager, webhook_event_manager
 from langflow.services.telemetry.schema import RunPayload
-from langflow.utils.compression import compress_response
 from langflow.utils.version import get_version_info
 
 if TYPE_CHECKING:
@@ -275,7 +274,7 @@ async def get_all(
         all_types = translate_component_dict(visible_types_en, locale) if locale != "en" else visible_types_en
 
         component_display_names = build_component_display_names(visible_types_en)
-        return compress_response({**all_types, "component_display_names": component_display_names})
+        return JSONResponse(content=jsonable_encoder({**all_types, "component_display_names": component_display_names}))
 
     except Exception as exc:
         raise HTTPException(status_code=500, detail=str(exc)) from exc

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

@@ -11,6 +11,7 @@
 import orjson
 from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile, status
 from fastapi.encoders import jsonable_encoder
+from fastapi.responses import JSONResponse
 from fastapi_pagination import Page, Params
 from fastapi_pagination.ext.sqlmodel import apaginate
 from lfx.log.logger import logger
@@ -106,7 +107,6 @@
 from langflow.services.database.models.user.model import User, UserRead
 from langflow.services.deps import get_catalog_policy_service, get_settings_service, get_storage_service
 from langflow.services.storage.service import StorageService
-from langflow.utils.compression import compress_response
 from langflow.utils.i18n import translate_flow_notes, translate_starter_flows
 
 # Re-export helpers so existing ``from langflow.api.v1.flows import ...`` still works.
@@ -447,13 +447,13 @@ async def read_flows(
                     act=FlowAction.READ,
                 )
             if header_flows:
-                # Convert to FlowHeader objects and compress the response
+                # Convert to FlowHeader objects
                 flow_headers = [FlowHeader.model_validate(flow, from_attributes=True) for flow in flows]
-                return compress_response(flow_headers)
+                return JSONResponse(content=jsonable_encoder(flow_headers))
 
             # 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)
+            return JSONResponse(content=jsonable_encoder(flow_reads))
 
         stmt = stmt.where(Flow.folder_id == folder_id)
 
@@ -741,8 +741,6 @@ async def upsert_flow(
 
     Returns 201 for creation, 200 for update.  Returns 404 if owned by another user.
     """
-    from fastapi.responses import JSONResponse
-
     # Read once, outside the retry loop: a rollback between attempts expires the ORM User
     # and a later attribute read would lazy-load outside the greenlet.
     writer_id = current_user.id
@@ -1381,7 +1379,7 @@ async def read_basic_examples(
                 blocked_template_keys=catalog_policy_snapshot.blocked_template_keys,
             )
         )
-        return compress_response(visible_flows)
+        return JSONResponse(content=jsonable_encoder(visible_flows))
 
     async with _starter_flows_lock:
         # Double-check inside lock to prevent thundering herd
@@ -1395,7 +1393,7 @@ async def read_basic_examples(
                     blocked_template_keys=catalog_policy_snapshot.blocked_template_keys,
                 )
             )
-            return compress_response(visible_flows)
+            return JSONResponse(content=jsonable_encoder(visible_flows))
 
         # Ensure raw DB data is cached
         cached_flow_reads = _starter_flows_cache.get("starter_flows")
@@ -1408,7 +1406,7 @@ async def read_basic_examples(
                 ).first()
 
                 if not starter_folder:
-                    return compress_response([])
+                    return JSONResponse(content=jsonable_encoder([]))
 
                 all_starter_folder_flows = (
                     await session.exec(select(Flow).where(Flow.folder_id == starter_folder.id))
@@ -1449,7 +1447,7 @@ async def read_basic_examples(
             blocked_template_keys=catalog_policy_snapshot.blocked_template_keys,
         )
     )
-    return compress_response(visible_flows)
+    return JSONResponse(content=jsonable_encoder(visible_flows))
 
 
 @router.post("/expand/", status_code=200, dependencies=[Depends(get_current_active_user)], include_in_schema=False)

src/backend/base/langflow/main.py

@@ -32,6 +32,7 @@
 from pydantic import PydanticDeprecatedSince20
 from pydantic_core import PydanticSerializationError
 from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
+from starlette.middleware.gzip import DEFAULT_EXCLUDED_CONTENT_TYPES, GZipMiddleware
 
 from langflow.api import log_router
 from langflow.api.health_check_router import health_check_router
@@ -77,6 +78,9 @@
 _tasks: list[asyncio.Task] = []
 
 MAX_PORT = 65535
+GZIP_MINIMUM_SIZE = 1000
+GZIP_COMPRESS_LEVEL = 6
+GZIP_EXCLUDED_CONTENT_TYPES = (*DEFAULT_EXCLUDED_CONTENT_TYPES, "application/octet-stream")
 
 # Enterprise lifespan hook registry. Enterprise plugins append async callables
 # at app-construction time (plugin registration runs before the lifespan
@@ -861,6 +865,12 @@ def create_app():
         lifespan=lifespan,
         root_path=settings.root_path,
     )
+    app.add_middleware(
+        GZipMiddleware,
+        minimum_size=GZIP_MINIMUM_SIZE,
+        compresslevel=GZIP_COMPRESS_LEVEL,
+        exclude_content_types=GZIP_EXCLUDED_CONTENT_TYPES,
+    )
     app.add_middleware(
         ContentSizeLimitMiddleware,
     )

src/backend/base/langflow/services/database/models/flow_version/model.py

@@ -6,7 +6,9 @@
 from pydantic import BaseModel, computed_field, field_serializer
 from pydantic import Field as PydanticField
 from sqlalchemy import CheckConstraint, Column, DateTime, ForeignKey, UniqueConstraint, func
-from sqlmodel import JSON, Field, SQLModel
+from sqlmodel import Field, SQLModel
+
+from langflow.services.database.models.flow_version.serialization import GzippedJSON
 
 
 class FlowVersion(SQLModel, table=True):  # type: ignore[call-arg]
@@ -20,7 +22,7 @@ class FlowVersion(SQLModel, table=True):  # type: ignore[call-arg]
     user_id: UUID | None = Field(
         sa_column=Column(ForeignKey("user.id", ondelete="SET NULL"), index=True, nullable=True),
     )
-    data: dict | None = Field(default=None, sa_column=Column(JSON))
+    data: dict | None = Field(default=None, sa_column=Column("data_gz", GzippedJSON))
     version_number: int = Field(nullable=False, ge=1)
     description: str | None = Field(default=None, nullable=True, max_length=500)
     created_at: datetime = Field(

src/backend/base/langflow/services/database/models/flow_version/serialization.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+import gzip
+import json
+from typing import TYPE_CHECKING, Any
+
+from sqlalchemy import LargeBinary
+from sqlalchemy.types import TypeDecorator
+
+from langflow.services.database.models.flow_version.exceptions import FlowVersionSerializationError
+
+if TYPE_CHECKING:
+    from sqlalchemy.engine.interfaces import Dialect
+
+COMPRESS_LEVEL = 6
+
+
+def pack(data: dict[str, Any] | None) -> bytes | None:
+    if data is None:
+        return None
+    try:
+        encoded = json.dumps(data, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
+    except (TypeError, ValueError) as exc:
+        msg = "Flow version data could not be serialized."
+        raise FlowVersionSerializationError(msg) from exc
+    return gzip.compress(encoded, COMPRESS_LEVEL)
+
+
+def unpack(blob: bytes | None) -> dict[str, Any] | None:
+    if blob is None:
+        return None
+    try:
+        return json.loads(gzip.decompress(blob))
+    except (OSError, EOFError, ValueError, UnicodeDecodeError) as exc:
+        msg = "Flow version data could not be read."
+        raise FlowVersionSerializationError(msg) from exc
+
+
+class GzippedJSON(TypeDecorator):
+    impl = LargeBinary
+    cache_ok = True
+
+    def process_bind_param(self, value: dict[str, Any] | None, dialect: Dialect) -> bytes | None:  # noqa: ARG002
+        return pack(value)
+
+    def process_result_value(self, value: bytes | None, dialect: Dialect) -> dict[str, Any] | None:  # noqa: ARG002
+        return unpack(value)

src/backend/base/pyproject.toml

@@ -19,6 +19,7 @@ maintainers = [
 dependencies = [
     "lfx~=1.13.0",
     "fastapi>=0.139.0,<1.0.0",
+    "starlette>=1.5.0",
     "slowapi>=0.1.9,<1.0.0",
     "httpx[http2]>=0.27,<1.0.0",
     "h2>=4.4.1",