refactor(api): migrate console recommended-app response to BaseModel (#35206)

Co-authored-by: ai-hpc <ai-hpc@users.noreply.github.com>
Co-authored-by: Asuka Minato <i@asukaminato.eu.org>

api/controllers/console/app/app.py

@@ -6,7 +6,6 @@ from typing import Any, Literal
 from flask import request
 from flask_restx import Resource
 from graphon.enums import WorkflowExecutionStatus
-from graphon.file import helpers as file_helpers
 from pydantic import AliasChoices, BaseModel, Field, computed_field, field_validator
 from sqlalchemy import select
 from sqlalchemy.orm import sessionmaker
@@ -31,6 +30,7 @@ from core.rag.retrieval.retrieval_methods import RetrievalMethod
 from core.trigger.constants import TRIGGER_NODE_TYPES
 from extensions.ext_database import db
 from fields.base import ResponseModel
+from libs.helper import build_icon_url
 from libs.login import current_account_with_tenant, login_required
 from models import App, DatasetPermissionEnum, Workflow
 from models.model import IconType
@@ -161,15 +161,6 @@ def _to_timestamp(value: datetime | int | None) -> int | None:
     return value
 
 
-def _build_icon_url(icon_type: str | IconType | None, icon: str | None) -> str | None:
-    if icon is None or icon_type is None:
-        return None
-    icon_type_value = icon_type.value if isinstance(icon_type, IconType) else str(icon_type)
-    if icon_type_value.lower() != IconType.IMAGE:
-        return None
-    return file_helpers.get_signed_file_url(icon)
-
-
 class Tag(ResponseModel):
     id: str
     name: str
@@ -292,7 +283,7 @@ class Site(ResponseModel):
     @computed_field(return_type=str | None)  # type: ignore
     @property
     def icon_url(self) -> str | None:
-        return _build_icon_url(self.icon_type, self.icon)
+        return build_icon_url(self.icon_type, self.icon)
 
     @field_validator("icon_type", mode="before")
     @classmethod
@@ -342,7 +333,7 @@ class AppPartial(ResponseModel):
     @computed_field(return_type=str | None)  # type: ignore
     @property
     def icon_url(self) -> str | None:
-        return _build_icon_url(self.icon_type, self.icon)
+        return build_icon_url(self.icon_type, self.icon)
 
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
@@ -390,7 +381,7 @@ class AppDetailWithSite(AppDetail):
     @computed_field(return_type=str | None)  # type: ignore
     @property
     def icon_url(self) -> str | None:
-        return _build_icon_url(self.icon_type, self.icon)
+        return build_icon_url(self.icon_type, self.icon)
 
 
 class AppPagination(ResponseModel):

api/controllers/console/explore/recommended_app.py

@@ -1,66 +1,83 @@
+from typing import Any
+
 from flask import request
-from flask_restx import Resource, fields, marshal_with
-from pydantic import BaseModel, Field
+from flask_restx import Resource
+from pydantic import BaseModel, Field, computed_field, field_validator
 
 from constants.languages import languages
-from controllers.common.schema import get_or_create_model
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required
-from libs.helper import AppIconUrlField
+from fields.base import ResponseModel
+from libs.helper import build_icon_url
 from libs.login import current_user, login_required
 from services.recommended_app_service import RecommendedAppService
 
-app_fields = {
-    "id": fields.String,
-    "name": fields.String,
-    "mode": fields.String,
-    "icon": fields.String,
-    "icon_type": fields.String,
-    "icon_url": AppIconUrlField,
-    "icon_background": fields.String,
-}
-
-app_model = get_or_create_model("RecommendedAppInfo", app_fields)
-
-recommended_app_fields = {
-    "app": fields.Nested(app_model, attribute="app"),
-    "app_id": fields.String,
-    "description": fields.String(attribute="description"),
-    "copyright": fields.String,
-    "privacy_policy": fields.String,
-    "custom_disclaimer": fields.String,
-    "category": fields.String,
-    "position": fields.Integer,
-    "is_listed": fields.Boolean,
-    "can_trial": fields.Boolean,
-}
-
-recommended_app_model = get_or_create_model("RecommendedApp", recommended_app_fields)
-
-recommended_app_list_fields = {
-    "recommended_apps": fields.List(fields.Nested(recommended_app_model)),
-    "categories": fields.List(fields.String),
-}
-
-recommended_app_list_model = get_or_create_model("RecommendedAppList", recommended_app_list_fields)
-
 
 class RecommendedAppsQuery(BaseModel):
     language: str | None = Field(default=None)
 
 
-console_ns.schema_model(
-    RecommendedAppsQuery.__name__,
-    RecommendedAppsQuery.model_json_schema(ref_template="#/definitions/{model}"),
+class RecommendedAppInfoResponse(ResponseModel):
+    id: str
+    name: str | None = None
+    mode: str | None = None
+    icon: str | None = None
+    icon_type: str | None = None
+    icon_background: str | None = None
+
+    @staticmethod
+    def _normalize_enum_like(value: Any) -> str | None:
+        if value is None:
+            return None
+        if isinstance(value, str):
+            return value
+        return str(getattr(value, "value", value))
+
+    @field_validator("mode", "icon_type", mode="before")
+    @classmethod
+    def _normalize_enum_fields(cls, value: Any) -> str | None:
+        return cls._normalize_enum_like(value)
+
+    @computed_field(return_type=str | None)  # type: ignore[prop-decorator]
+    @property
+    def icon_url(self) -> str | None:
+        return build_icon_url(self.icon_type, self.icon)
+
+
+class RecommendedAppResponse(ResponseModel):
+    app: RecommendedAppInfoResponse | None = None
+    app_id: str
+    description: str | None = None
+    copyright: str | None = None
+    privacy_policy: str | None = None
+    custom_disclaimer: str | None = None
+    category: str | None = None
+    position: int | None = None
+    is_listed: bool | None = None
+    can_trial: bool | None = None
+
+
+class RecommendedAppListResponse(ResponseModel):
+    recommended_apps: list[RecommendedAppResponse]
+    categories: list[str]
+
+
+register_schema_models(
+    console_ns,
+    RecommendedAppsQuery,
+    RecommendedAppInfoResponse,
+    RecommendedAppResponse,
+    RecommendedAppListResponse,
 )
 
 
 @console_ns.route("/explore/apps")
 class RecommendedAppListApi(Resource):
     @console_ns.expect(console_ns.models[RecommendedAppsQuery.__name__])
+    @console_ns.response(200, "Success", console_ns.models[RecommendedAppListResponse.__name__])
     @login_required
     @account_initialization_required
-    @marshal_with(recommended_app_list_model)
     def get(self):
         # language args
         args = RecommendedAppsQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
@@ -72,7 +89,10 @@ class RecommendedAppListApi(Resource):
         else:
             language_prefix = languages[0]
 
-        return RecommendedAppService.get_recommended_apps_and_categories(language_prefix)
+        return RecommendedAppListResponse.model_validate(
+            RecommendedAppService.get_recommended_apps_and_categories(language_prefix),
+            from_attributes=True,
+        ).model_dump(mode="json")
 
 
 @console_ns.route("/explore/apps/<uuid:app_id>")

api/libs/helper.py

@@ -120,10 +120,22 @@ class AppIconUrlField(fields.Raw):
             obj = obj["app"]
 
         if isinstance(obj, App | Site) and obj.icon_type == IconType.IMAGE:
-            return file_helpers.get_signed_file_url(obj.icon)
+            return build_icon_url(obj.icon_type, obj.icon)
         return None
 
 
+def build_icon_url(icon_type: Any, icon: str | None) -> str | None:
+    if icon is None or icon_type is None:
+        return None
+
+    from models.model import IconType
+
+    icon_type_value = icon_type.value if isinstance(icon_type, IconType) else str(icon_type)
+    if icon_type_value.lower() != IconType.IMAGE:
+        return None
+    return file_helpers.get_signed_file_url(icon)
+
+
 class AvatarUrlField(fields.Raw):
     def output(self, key, obj, **kwargs):
         if obj is None:

api/tests/unit_tests/controllers/console/app/test_app_response_models.py

@@ -138,12 +138,15 @@ def app_models(app_module):
 def patch_signed_url(monkeypatch, app_module):
     """Ensure icon URL generation uses a deterministic helper for tests."""
 
-    def _fake_signed_url(key: str | None) -> str | None:
-        if not key:
+    def _fake_build_icon_url(_icon_type, key: str | None) -> str | None:
+        if key is None:
+            return None
+        icon_type = str(_icon_type).lower()
+        if icon_type != "image":
             return None
         return f"signed:{key}"
 
-    monkeypatch.setattr(app_module.file_helpers, "get_signed_file_url", _fake_signed_url)
+    monkeypatch.setattr(app_module, "build_icon_url", _fake_build_icon_url)
 
 
 def _ts(hour: int = 12) -> datetime:

api/tests/unit_tests/controllers/console/explore/test_recommended_app.py

@@ -1,6 +1,7 @@
 from unittest.mock import MagicMock, patch
 
 import controllers.console.explore.recommended_app as module
+from models.model import AppMode, IconType
 
 
 def unwrap(func):
@@ -90,3 +91,48 @@ class TestRecommendedAppApi:
 
         service_mock.assert_called_once_with("11111111-1111-1111-1111-111111111111")
         assert result == result_data
+
+
+class TestRecommendedAppResponseModels:
+    def test_recommended_app_info_response_computes_icon_url(self):
+        with patch.object(module, "build_icon_url", return_value="https://signed/icon.png"):
+            payload = module.RecommendedAppInfoResponse.model_validate(
+                {
+                    "id": "app-1",
+                    "name": "App",
+                    "mode": AppMode.CHAT,
+                    "icon": "icon.png",
+                    "icon_type": IconType.IMAGE,
+                    "icon_background": "#fff",
+                }
+            ).model_dump(mode="json")
+
+        assert payload["icon_url"] == "https://signed/icon.png"
+
+    def test_recommended_app_list_response_serialization(self):
+        response = module.RecommendedAppListResponse.model_validate(
+            {
+                "recommended_apps": [
+                    {
+                        "app": {
+                            "id": "app-1",
+                            "name": "App",
+                            "mode": "chat",
+                            "icon": "icon.png",
+                            "icon_type": "emoji",
+                            "icon_background": "#fff",
+                        },
+                        "app_id": "app-1",
+                        "description": "desc",
+                        "category": "cat",
+                        "position": 1,
+                        "is_listed": True,
+                        "can_trial": False,
+                    }
+                ],
+                "categories": ["cat"],
+            }
+        ).model_dump(mode="json")
+
+        assert response["recommended_apps"][0]["app_id"] == "app-1"
+        assert response["categories"] == ["cat"]