perf(api): compress flow payloads on the wire

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,16 @@
 _tasks: list[asyncio.Task] = []
 
 MAX_PORT = 65535
+GZIP_MINIMUM_SIZE = 1000
+GZIP_COMPRESS_LEVEL = 6
+GZIP_ALREADY_COMPRESSED_CONTENT_TYPES = (
+    "application/octet-stream",
+    "application/pdf",
+    "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
+    "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+    "application/vnd.openxmlformats-officedocument.presentationml.presentation",
+)
+GZIP_EXCLUDED_CONTENT_TYPES = (*DEFAULT_EXCLUDED_CONTENT_TYPES, *GZIP_ALREADY_COMPRESSED_CONTENT_TYPES)
 
 # Enterprise lifespan hook registry. Enterprise plugins append async callables
 # at app-construction time (plugin registration runs before the lifespan
@@ -861,6 +872,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/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",

src/backend/tests/unit/api/test_response_compression.py

@@ -0,0 +1,135 @@
+import json
+
+import pytest
+from fastapi import FastAPI, Response
+from httpx import ASGITransport, AsyncClient
+from langflow.main import GZIP_COMPRESS_LEVEL, GZIP_EXCLUDED_CONTENT_TYPES, GZIP_MINIMUM_SIZE
+from starlette.middleware.gzip import GZipMiddleware
+
+LARGE_NODE_COUNT = 40
+
+
+def _large_flow_payload() -> dict:
+    nodes = [
+        {
+            "id": f"node-{index}",
+            "data": {"node": {"template": {"code": {"value": "from lfx.custom import Component\n" * 12}}}},
+        }
+        for index in range(LARGE_NODE_COUNT)
+    ]
+    return {
+        "name": "compression fixture",
+        "description": "flow large enough to cross the compression threshold",
+        "data": {"nodes": nodes, "edges": []},
+        "is_component": False,
+        "webhook": False,
+    }
+
+
+async def _create_large_flow(client: AsyncClient, headers: dict) -> str:
+    payload = _large_flow_payload()
+    assert len(json.dumps(payload).encode()) > GZIP_MINIMUM_SIZE
+    response = await client.post("api/v1/flows/", json=payload, headers=headers)
+    assert response.status_code == 201
+    return response.json()["id"]
+
+
+async def test_flow_read_is_compressed_when_the_client_accepts_gzip(client: AsyncClient, logged_in_headers):
+    flow_id = await _create_large_flow(client, logged_in_headers)
+
+    response = await client.get(f"api/v1/flows/{flow_id}", headers={**logged_in_headers, "Accept-Encoding": "gzip"})
+
+    assert response.status_code == 200
+    assert response.headers["content-encoding"] == "gzip"
+    assert "accept-encoding" in response.headers["vary"].lower()
+    assert response.json()["id"] == flow_id
+
+
+async def test_flow_read_is_untouched_when_the_client_does_not_accept_gzip(client: AsyncClient, logged_in_headers):
+    flow_id = await _create_large_flow(client, logged_in_headers)
+
+    response = await client.get(f"api/v1/flows/{flow_id}", headers={**logged_in_headers, "Accept-Encoding": "identity"})
+
+    assert response.status_code == 200
+    assert "content-encoding" not in response.headers
+    assert json.loads(response.content)["id"] == flow_id
+
+
+async def test_flow_update_echo_is_compressed(client: AsyncClient, logged_in_headers):
+    flow_id = await _create_large_flow(client, logged_in_headers)
+    payload = _large_flow_payload()
+    payload["name"] = "compression fixture renamed"
+
+    response = await client.patch(
+        f"api/v1/flows/{flow_id}",
+        json=payload,
+        headers={**logged_in_headers, "Accept-Encoding": "gzip"},
+    )
+
+    assert response.status_code == 200
+    assert response.headers["content-encoding"] == "gzip"
+    assert response.json()["name"] == "compression fixture renamed"
+
+
+async def test_response_below_the_threshold_is_not_compressed(client: AsyncClient, logged_in_headers):
+    response = await client.get("api/v1/version", headers={**logged_in_headers, "Accept-Encoding": "gzip"})
+
+    assert response.status_code == 200
+    assert len(response.content) < GZIP_MINIMUM_SIZE
+    assert "content-encoding" not in response.headers
+
+
+EXCLUDED_UNDER_TEST = (
+    "application/octet-stream",
+    "application/zip",
+    "text/event-stream",
+    "image/png",
+    "application/pdf",
+    "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
+    "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+    "application/vnd.openxmlformats-officedocument.presentationml.presentation",
+)
+
+
+def _app_with_the_same_middleware() -> FastAPI:
+    app = FastAPI()
+    app.add_middleware(
+        GZipMiddleware,
+        minimum_size=GZIP_MINIMUM_SIZE,
+        compresslevel=GZIP_COMPRESS_LEVEL,
+        exclude_content_types=GZIP_EXCLUDED_CONTENT_TYPES,
+    )
+
+    @app.get("/payload")
+    async def payload(content_type: str) -> Response:
+        return Response(content=b"x" * (GZIP_MINIMUM_SIZE * 4), media_type=content_type)
+
+    return app
+
+
+def test_binary_and_streaming_content_types_are_configured_as_excluded():
+    for content_type in EXCLUDED_UNDER_TEST:
+        assert content_type in GZIP_EXCLUDED_CONTENT_TYPES
+
+
+@pytest.mark.parametrize("content_type", EXCLUDED_UNDER_TEST)
+async def test_excluded_content_types_are_not_compressed(content_type):
+    transport = ASGITransport(app=_app_with_the_same_middleware())
+    async with AsyncClient(transport=transport, base_url="http://testserver") as client:
+        response = await client.get(
+            "/payload", params={"content_type": content_type}, headers={"Accept-Encoding": "gzip"}
+        )
+
+    assert response.status_code == 200
+    assert "content-encoding" not in response.headers
+    assert len(response.content) == GZIP_MINIMUM_SIZE * 4
+
+
+async def test_a_compressible_type_on_the_same_app_is_compressed():
+    transport = ASGITransport(app=_app_with_the_same_middleware())
+    async with AsyncClient(transport=transport, base_url="http://testserver") as client:
+        response = await client.get(
+            "/payload", params={"content_type": "application/json"}, headers={"Accept-Encoding": "gzip"}
+        )
+
+    assert response.headers["content-encoding"] == "gzip"