perf(api): defer workflow node factory imports

api/controllers/common/schema.py

@@ -7,7 +7,9 @@ These helpers keep that translation centralized so models registered through
 """
 
 from collections.abc import Iterable, Mapping
+from copy import deepcopy
 from enum import StrEnum
+from functools import lru_cache
 from typing import Any, Literal, NotRequired, Protocol, TypedDict
 
 from flask import request
@@ -72,11 +74,22 @@ def _register_json_schema(namespace: Namespace, name: str, schema: dict) -> None
 JsonSchemaMode = Literal["validation", "serialization"]
 
 
+@lru_cache(maxsize=256)
+def _model_json_schema(model: type[BaseModel], mode: JsonSchemaMode, _validator: object) -> dict[str, Any]:
+    """Reuse schema generation across namespaces, invalidating on model rebuild.
+
+    Pydantic replaces its validator when rebuilding a model. Including that
+    object in the cache key keeps registrations after a rebuild up to date.
+    Callers must copy the result before exposing it to mutable RESTX models.
+    """
+    return model.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_OPENAPI_3_0, mode=mode)
+
+
 def _register_schema_model(namespace: Namespace, model: type[BaseModel], *, mode: JsonSchemaMode) -> None:
     _register_json_schema(
         namespace,
         model.__name__,
-        model.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_OPENAPI_3_0, mode=mode),
+        deepcopy(_model_json_schema(model, mode, model.__pydantic_validator__)),
     )
 
 

api/core/app/workflow/__init__.py

@@ -1,3 +1,19 @@
-from core.workflow.node_factory import DifyNodeFactory
+"""Workflow app adapters; importing file helpers does not load every node runtime."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.workflow.node_factory import DifyNodeFactory
 
 __all__ = ["DifyNodeFactory"]
+
+
+def __getattr__(name: str) -> type[DifyNodeFactory]:
+    if name == "DifyNodeFactory":
+        from core.workflow.node_factory import DifyNodeFactory
+
+        globals()[name] = DifyNodeFactory
+        return DifyNodeFactory
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

api/tests/unit_tests/conftest.py

@@ -2,9 +2,11 @@ from __future__ import annotations
 
 import os
 import shutil
+import sys
 from collections.abc import Callable, Iterator
 from pathlib import Path
 from tempfile import TemporaryDirectory
+from types import ModuleType
 from typing import TYPE_CHECKING
 from unittest.mock import MagicMock, patch
 
@@ -53,16 +55,20 @@ if TYPE_CHECKING:
 
 
 def _patch_redis_clients_on_loaded_modules() -> None:
-    """Ensure any module-level redis_client references point to the shared redis_mock."""
+    """Rebind concrete module globals without invoking module-level ``__getattr__``.
 
-    import sys
-
-    for module in list(sys.modules.values()):
-        if module is None:
+    Inspect each time so late imports, reloads, and newly added globals are still
+    covered. Attribute lookup on thousands of unrelated lazy modules is costly
+    and can itself trigger imports; only stored references need rebinding.
+    """
+    for module in sys.modules.copy().values():
+        if not isinstance(module, ModuleType):
             continue
-        for client_attribute in ("redis_client", "_pubsub_redis_client"):
-            if hasattr(module, client_attribute):
-                setattr(module, client_attribute, redis_mock)
+        namespace = module.__dict__
+        if "redis_client" in namespace:
+            namespace["redis_client"] = redis_mock
+        if "_pubsub_redis_client" in namespace:
+            namespace["_pubsub_redis_client"] = redis_mock
 
 
 @pytest.fixture

api/tests/unit_tests/controllers/common/test_schema.py

@@ -132,6 +132,52 @@ def test_register_schema_model_promotes_nested_pydantic_definitions():
     assert called_schemas["ChildModel"] == parent_schema["$defs"]["ChildModel"]
 
 
+def test_schema_generation_is_reused_across_namespaces_but_separates_modes():
+    from controllers.common.schema import register_response_schema_model, register_schema_model
+
+    class SharedModel(BaseModel):
+        public_name: str = Field(validation_alias="internal_name")
+
+    with patch.object(SharedModel, "model_json_schema", wraps=SharedModel.model_json_schema) as generate:
+        request_namespace = Namespace("request")
+        response_namespace = Namespace("response")
+        register_schema_model(request_namespace, SharedModel)
+        register_schema_model(Namespace("other-request"), SharedModel)
+        register_response_schema_model(response_namespace, SharedModel)
+
+    assert generate.call_count == 2
+    assert "internal_name" in request_namespace.models["SharedModel"].__schema__["properties"]
+    assert "public_name" in response_namespace.models["SharedModel"].__schema__["properties"]
+
+
+def test_cached_schema_does_not_share_mutable_nested_definitions():
+    from controllers.common.schema import register_schema_model
+
+    first = Namespace("first")
+    register_schema_model(first, ParentModel)
+    first.models["ChildModel"].__schema__["properties"]["value"]["description"] = "changed"
+    first.models["ParentModel"].__schema__["properties"]["child"]["$ref"] = "changed"
+
+    second = Namespace("second")
+    register_schema_model(second, ParentModel)
+    assert "description" not in second.models["ChildModel"].__schema__["properties"]["value"]
+    assert second.models["ParentModel"].__schema__["properties"]["child"]["$ref"] == "#/components/schemas/ChildModel"
+
+
+def test_schema_cache_refreshes_after_model_rebuild():
+    from controllers.common.schema import register_schema_model
+
+    class RebuiltModel(BaseModel):
+        value: str = Field(description="before")
+
+    namespace = Namespace("rebuilt")
+    register_schema_model(namespace, RebuiltModel)
+    RebuiltModel.model_fields["value"].description = "after"
+    RebuiltModel.model_rebuild(force=True)
+    register_schema_model(namespace, RebuiltModel)
+    assert namespace.models["RebuiltModel"].__schema__["properties"]["value"]["description"] == "after"
+
+
 def test_register_schema_models_registers_multiple_models():
     from controllers.common.schema import register_schema_models
 

api/tests/unit_tests/controllers/conftest.py

@@ -1,6 +1,9 @@
 """Explicit App service composition for controller tests that use app queries."""
 
+import os
+from collections.abc import Callable
 from dataclasses import dataclass
+from functools import cache
 
 import pytest
 from flask import Flask
@@ -20,6 +23,42 @@ from services.app.import_service import AppImportService
 from services.app.query_service import AppQueryService
 from services.tag_application_service import TagApplicationService
 from services.webapp_access_query_service import WebAppAccessQueryService
+from tests.unit_tests.config_override import config_overrides_context
+
+
+@pytest.fixture(scope="session")
+def _console_spec_loader(tmp_path_factory: pytest.TempPathFactory) -> Callable[[], str]:
+    """Export once on demand; consumers parse independent copies of the JSON."""
+
+    @cache
+    def load() -> str:
+        from configs import dify_config
+        from dev.generate_swagger_specs import generate_specs
+
+        output_dir = tmp_path_factory.mktemp("controller-specs")
+        # The exporter sets generation-only defaults. Restore them even if the
+        # first consumer has different per-test config or generation fails.
+        original_config = dify_config.model_dump(
+            include={"SECRET_KEY", "STORAGE_TYPE", "STORAGE_LOCAL_PATH", "SWAGGER_UI_ENABLED"}
+        )
+        with config_overrides_context(**original_config), pytest.MonkeyPatch.context() as monkeypatch:
+            for name, default in (
+                ("SECRET_KEY", "spec-export"),
+                ("STORAGE_TYPE", "local"),
+                ("STORAGE_LOCAL_PATH", "/tmp/dify-storage"),
+            ):
+                monkeypatch.setenv(name, os.environ.get(name, default))
+            written_paths = generate_specs(output_dir)
+        console_path = output_dir / "console-openapi.json"
+        assert console_path in written_paths
+        return console_path.read_text(encoding="utf-8")
+
+    return load
+
+
+@pytest.fixture
+def exported_console_json(_console_spec_loader: Callable[[], str]) -> str:
+    return _console_spec_loader()
 
 
 @dataclass

api/tests/unit_tests/controllers/test_rbac_access_policy_schema.py

@@ -15,13 +15,9 @@ def _object_at(root: dict[str, object], *keys: str) -> dict[str, object]:
     return cast(dict[str, object], current)
 
 
-@pytest.fixture(scope="module")
-def console_schema(tmp_path_factory: pytest.TempPathFactory) -> dict[str, object]:
-    from dev.generate_swagger_specs import generate_specs
-
-    output_dir = tmp_path_factory.mktemp("rbac-openapi")
-    generate_specs(output_dir)
-    schema: object = json.loads((output_dir / "console-openapi.json").read_text())
+@pytest.fixture
+def console_schema(exported_console_json: str) -> dict[str, object]:
+    schema: object = json.loads(exported_console_json)
     assert isinstance(schema, dict)
     return cast(dict[str, object], schema)
 

api/tests/unit_tests/controllers/test_swagger.py

@@ -2,7 +2,6 @@
 
 import json
 from collections.abc import Callable, Iterator
-from pathlib import Path
 
 import pytest
 from flask import Blueprint, Flask
@@ -41,23 +40,6 @@ def openapi_json() -> OpenAPIJSONLoader:
     return load
 
 
-@pytest.fixture(scope="module")
-def _exported_spec_directory(tmp_path_factory: pytest.TempPathFactory) -> Path:
-    return tmp_path_factory.mktemp("swagger-specs")
-
-
-@pytest.fixture
-def exported_console_json(_exported_spec_directory: Path, _swagger_config: None) -> str:
-    """Exercise the exporter once; tests parse independent copies of its output."""
-    from dev.generate_swagger_specs import generate_specs
-
-    console_openapi_path = _exported_spec_directory / "console-openapi.json"
-    if not console_openapi_path.exists():
-        written_paths = generate_specs(_exported_spec_directory)
-        assert console_openapi_path in written_paths
-    return console_openapi_path.read_text(encoding="utf-8")
-
-
 USER_PROPERTY_SCHEMA = {
     "description": (
         "End-user identifier, defined by your app and unique within it. Identifies the end user for this request. "

api/tests/unit_tests/core/app/workflow/test_package.py

@@ -0,0 +1,27 @@
+import subprocess
+import sys
+from pathlib import Path
+
+
+def test_workflow_package_defers_node_factory_and_preserves_its_export() -> None:
+    """Use a fresh interpreter so collection's imports cannot hide eager loading."""
+    result = subprocess.run(
+        [
+            sys.executable,
+            "-c",
+            "import sys\n"
+            "import core.app.workflow as workflow\n"
+            "assert 'core.workflow.node_factory' not in sys.modules\n"
+            "from core.app.workflow import DifyNodeFactory\n"
+            "from core.workflow.node_factory import DifyNodeFactory as factory\n"
+            "assert DifyNodeFactory is factory\n"
+            "assert workflow.DifyNodeFactory is factory\n"
+            "assert not hasattr(workflow, 'missing_export')\n",
+        ],
+        cwd=Path(__file__).resolve().parents[5],
+        capture_output=True,
+        text=True,
+        timeout=20,
+        check=False,
+    )
+    assert result.returncode == 0, result.stderr

api/tests/unit_tests/test_redis_fixtures.py

@@ -0,0 +1,43 @@
+import sys
+from types import ModuleType
+
+import pytest
+
+from tests.unit_tests.conftest import _patch_redis_clients_on_loaded_modules, redis_mock
+
+
+@pytest.mark.parametrize("attribute", ["redis_client", "_pubsub_redis_client"])
+def test_redis_rebinding_covers_new_globals_and_replaced_modules(
+    monkeypatch: pytest.MonkeyPatch, attribute: str
+) -> None:
+    name = "_redis_fixture_probe"
+    module = ModuleType(name)
+    monkeypatch.setitem(sys.modules, name, module)
+    _patch_redis_clients_on_loaded_modules()
+
+    # A module can gain a reference after it was already seen by another test.
+    setattr(module, attribute, object())
+    _patch_redis_clients_on_loaded_modules()
+    assert vars(module)[attribute] is redis_mock
+
+    # Rebind on every test, even when the set of imported modules is unchanged.
+    setattr(module, attribute, object())
+    _patch_redis_clients_on_loaded_modules()
+    assert vars(module)[attribute] is redis_mock
+
+    replacement = ModuleType(name)
+    setattr(replacement, attribute, object())
+    monkeypatch.setitem(sys.modules, name, replacement)
+    _patch_redis_clients_on_loaded_modules()
+    assert vars(replacement)[attribute] is redis_mock
+
+
+def test_redis_rebinding_does_not_resolve_lazy_module_attributes(monkeypatch: pytest.MonkeyPatch) -> None:
+    module = ModuleType("_lazy_redis_fixture_probe")
+
+    def resolve(name: str) -> object:
+        pytest.fail(f"Redis rebinding must not import unrelated lazy attributes: {name}")
+
+    vars(module)["__getattr__"] = resolve
+    monkeypatch.setitem(sys.modules, module.__name__, module)
+    _patch_redis_clients_on_loaded_modules()