refactor: Refactor the service of retrieval the recommend app (#9302)

api/services/recommend_app/buildin/buildin_retrieval.py

+import json
+from os import path
+from pathlib import Path
+from typing import Optional
+
+from flask import current_app
+
+from services.recommend_app.recommend_app_base import RecommendAppRetrievalBase
+from services.recommend_app.recommend_app_type import RecommendAppType
+
+
+class BuildInRecommendAppRetrieval(RecommendAppRetrievalBase):
+    """
+    Retrieval recommended app from buildin, the location  is constants/recommended_apps.json
+    """
+
+    builtin_data: Optional[dict] = None
+
+    def get_type(self) -> str:
+        return RecommendAppType.BUILDIN
+
+    def get_recommended_apps_and_categories(self, language: str) -> dict:
+        result = self.fetch_recommended_apps_from_builtin(language)
+        return result
+
+    def get_recommend_app_detail(self, app_id: str):
+        result = self.fetch_recommended_app_detail_from_builtin(app_id)
+        return result
+
+    @classmethod
+    def _get_builtin_data(cls) -> dict:
+        """
+        Get builtin data.
+        :return:
+        """
+        if cls.builtin_data:
+            return cls.builtin_data
+
+        root_path = current_app.root_path
+        cls.builtin_data = json.loads(
+            Path(path.join(root_path, "constants", "recommended_apps.json")).read_text(encoding="utf-8")
+        )
+
+        return cls.builtin_data
+
+    @classmethod
+    def fetch_recommended_apps_from_builtin(cls, language: str) -> dict:
+        """
+        Fetch recommended apps from builtin.
+        :param language: language
+        :return:
+        """
+        builtin_data = cls._get_builtin_data()
+        return builtin_data.get("recommended_apps", {}).get(language)
+
+    @classmethod
+    def fetch_recommended_app_detail_from_builtin(cls, app_id: str) -> Optional[dict]:
+        """
+        Fetch recommended app detail from builtin.
+        :param app_id: App ID
+        :return:
+        """
+        builtin_data = cls._get_builtin_data()
+        return builtin_data.get("app_details", {}).get(app_id)

api/services/recommend_app/database/database_retrieval.py

+from typing import Optional
+
+from constants.languages import languages
+from extensions.ext_database import db
+from models.model import App, RecommendedApp
+from services.app_dsl_service import AppDslService
+from services.recommend_app.recommend_app_base import RecommendAppRetrievalBase
+from services.recommend_app.recommend_app_type import RecommendAppType
+
+
+class DatabaseRecommendAppRetrieval(RecommendAppRetrievalBase):
+    """
+    Retrieval recommended app from database
+    """
+
+    def get_recommended_apps_and_categories(self, language: str) -> dict:
+        result = self.fetch_recommended_apps_from_db(language)
+        return result
+
+    def get_recommend_app_detail(self, app_id: str):
+        result = self.fetch_recommended_app_detail_from_db(app_id)
+        return result
+
+    def get_type(self) -> str:
+        return RecommendAppType.DATABASE
+
+    @classmethod
+    def fetch_recommended_apps_from_db(cls, language: str) -> dict:
+        """
+        Fetch recommended apps from db.
+        :param language: language
+        :return:
+        """
+        recommended_apps = (
+            db.session.query(RecommendedApp)
+            .filter(RecommendedApp.is_listed == True, RecommendedApp.language == language)
+            .all()
+        )
+
+        if len(recommended_apps) == 0:
+            recommended_apps = (
+                db.session.query(RecommendedApp)
+                .filter(RecommendedApp.is_listed == True, RecommendedApp.language == languages[0])
+                .all()
+            )
+
+        categories = set()
+        recommended_apps_result = []
+        for recommended_app in recommended_apps:
+            app = recommended_app.app
+            if not app or not app.is_public:
+                continue
+
+            site = app.site
+            if not site:
+                continue
+
+            recommended_app_result = {
+                "id": recommended_app.id,
+                "app": {
+                    "id": app.id,
+                    "name": app.name,
+                    "mode": app.mode,
+                    "icon": app.icon,
+                    "icon_background": app.icon_background,
+                },
+                "app_id": recommended_app.app_id,
+                "description": site.description,
+                "copyright": site.copyright,
+                "privacy_policy": site.privacy_policy,
+                "custom_disclaimer": site.custom_disclaimer,
+                "category": recommended_app.category,
+                "position": recommended_app.position,
+                "is_listed": recommended_app.is_listed,
+            }
+            recommended_apps_result.append(recommended_app_result)
+
+            categories.add(recommended_app.category)
+
+        return {"recommended_apps": recommended_apps_result, "categories": sorted(categories)}
+
+    @classmethod
+    def fetch_recommended_app_detail_from_db(cls, app_id: str) -> Optional[dict]:
+        """
+        Fetch recommended app detail from db.
+        :param app_id: App ID
+        :return:
+        """
+        # is in public recommended list
+        recommended_app = (
+            db.session.query(RecommendedApp)
+            .filter(RecommendedApp.is_listed == True, RecommendedApp.app_id == app_id)
+            .first()
+        )
+
+        if not recommended_app:
+            return None
+
+        # get app detail
+        app_model = db.session.query(App).filter(App.id == app_id).first()
+        if not app_model or not app_model.is_public:
+            return None
+
+        return {
+            "id": app_model.id,
+            "name": app_model.name,
+            "icon": app_model.icon,
+            "icon_background": app_model.icon_background,
+            "mode": app_model.mode,
+            "export_data": AppDslService.export_dsl(app_model=app_model),
+        }

api/services/recommend_app/recommend_app_base.py

+from abc import ABC, abstractmethod
+
+
+class RecommendAppRetrievalBase(ABC):
+    """Interface for recommend app retrieval."""
+
+    @abstractmethod
+    def get_recommended_apps_and_categories(self, language: str) -> dict:
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_recommend_app_detail(self, app_id: str):
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_type(self) -> str:
+        raise NotImplementedError

api/services/recommend_app/recommend_app_factory.py

+from services.recommend_app.buildin.buildin_retrieval import BuildInRecommendAppRetrieval
+from services.recommend_app.database.database_retrieval import DatabaseRecommendAppRetrieval
+from services.recommend_app.recommend_app_base import RecommendAppRetrievalBase
+from services.recommend_app.recommend_app_type import RecommendAppType
+from services.recommend_app.remote.remote_retrieval import RemoteRecommendAppRetrieval
+
+
+class RecommendAppRetrievalFactory:
+    @staticmethod
+    def get_recommend_app_factory(mode: str) -> type[RecommendAppRetrievalBase]:
+        match mode:
+            case RecommendAppType.REMOTE:
+                return RemoteRecommendAppRetrieval
+            case RecommendAppType.DATABASE:
+                return DatabaseRecommendAppRetrieval
+            case RecommendAppType.BUILDIN:
+                return BuildInRecommendAppRetrieval
+            case _:
+                raise ValueError(f"invalid fetch recommended apps mode: {mode}")
+
+    @staticmethod
+    def get_buildin_recommend_app_retrieval():
+        return BuildInRecommendAppRetrieval

api/services/recommend_app/recommend_app_type.py

+from enum import Enum
+
+
+class RecommendAppType(str, Enum):
+    REMOTE = "remote"
+    BUILDIN = "builtin"
+    DATABASE = "db"

api/services/recommend_app/remote/remote_retrieval.py

+import logging
+from typing import Optional
+
+import requests
+
+from configs import dify_config
+from services.recommend_app.buildin.buildin_retrieval import BuildInRecommendAppRetrieval
+from services.recommend_app.recommend_app_base import RecommendAppRetrievalBase
+from services.recommend_app.recommend_app_type import RecommendAppType
+
+logger = logging.getLogger(__name__)
+
+
+class RemoteRecommendAppRetrieval(RecommendAppRetrievalBase):
+    """
+    Retrieval recommended app from dify official
+    """
+
+    def get_recommend_app_detail(self, app_id: str):
+        try:
+            result = self.fetch_recommended_app_detail_from_dify_official(app_id)
+        except Exception as e:
+            logger.warning(f"fetch recommended app detail from dify official failed: {e}, switch to built-in.")
+            result = BuildInRecommendAppRetrieval.fetch_recommended_app_detail_from_builtin(app_id)
+        return result
+
+    def get_recommended_apps_and_categories(self, language: str) -> dict:
+        try:
+            result = self.fetch_recommended_apps_from_dify_official(language)
+        except Exception as e:
+            logger.warning(f"fetch recommended apps from dify official failed: {e}, switch to built-in.")
+            result = BuildInRecommendAppRetrieval.fetch_recommended_apps_from_builtin(language)
+        return result
+
+    def get_type(self) -> str:
+        return RecommendAppType.REMOTE
+
+    @classmethod
+    def fetch_recommended_app_detail_from_dify_official(cls, app_id: str) -> Optional[dict]:
+        """
+        Fetch recommended app detail from dify official.
+        :param app_id: App ID
+        :return:
+        """
+        domain = dify_config.HOSTED_FETCH_APP_TEMPLATES_REMOTE_DOMAIN
+        url = f"{domain}/apps/{app_id}"
+        response = requests.get(url, timeout=(3, 10))
+        if response.status_code != 200:
+            return None
+
+        return response.json()
+
+    @classmethod
+    def fetch_recommended_apps_from_dify_official(cls, language: str) -> dict:
+        """
+        Fetch recommended apps from dify official.
+        :param language: language
+        :return:
+        """
+        domain = dify_config.HOSTED_FETCH_APP_TEMPLATES_REMOTE_DOMAIN
+        url = f"{domain}/apps?language={language}"
+        response = requests.get(url, timeout=(3, 10))
+        if response.status_code != 200:
+            raise ValueError(f"fetch recommended apps failed, status code: {response.status_code}")
+
+        result = response.json()
+
+        if "categories" in result:
+            result["categories"] = sorted(result["categories"])
+
+        return result

api/services/recommended_app_service.py

-import json
-import logging
-from os import path
-from pathlib import Path
 from typing import Optional
 
-import requests
-from flask import current_app
-
 from configs import dify_config
-from constants.languages import languages
-from extensions.ext_database import db
-from models.model import App, RecommendedApp
-from services.app_dsl_service import AppDslService
-
-logger = logging.getLogger(__name__)
+from services.recommend_app.recommend_app_factory import RecommendAppRetrievalFactory
 
 
 class RecommendedAppService:
-    builtin_data: Optional[dict] = None
-
     @classmethod
     def get_recommended_apps_and_categories(cls, language: str) -> dict:
         """
         :return:
         """
         mode = dify_config.HOSTED_FETCH_APP_TEMPLATES_MODE
-        if mode == "remote":
-            try:
-                result = cls._fetch_recommended_apps_from_dify_official(language)
-            except Exception as e:
-                logger.warning(f"fetch recommended apps from dify official failed: {e}, switch to built-in.")
-                result = cls._fetch_recommended_apps_from_builtin(language)
-        elif mode == "db":
-            result = cls._fetch_recommended_apps_from_db(language)
-        elif mode == "builtin":
-            result = cls._fetch_recommended_apps_from_builtin(language)
-        else:
-            raise ValueError(f"invalid fetch recommended apps mode: {mode}")
-
+        retrieval_instance = RecommendAppRetrievalFactory.get_recommend_app_factory(mode)()
+        result = retrieval_instance.get_recommended_apps_and_categories(language)
         if not result.get("recommended_apps") and language != "en-US":
-            result = cls._fetch_recommended_apps_from_builtin("en-US")
-
-        return result
-
-    @classmethod
-    def _fetch_recommended_apps_from_db(cls, language: str) -> dict:
-        """
-        Fetch recommended apps from db.
-        :param language: language
-        :return:
-        """
-        recommended_apps = (
-            db.session.query(RecommendedApp)
-            .filter(RecommendedApp.is_listed == True, RecommendedApp.language == language)
-            .all()
-        )
-
-        if len(recommended_apps) == 0:
-            recommended_apps = (
-                db.session.query(RecommendedApp)
-                .filter(RecommendedApp.is_listed == True, RecommendedApp.language == languages[0])
-                .all()
+            result = (
+                RecommendAppRetrievalFactory.get_buildin_recommend_app_retrieval().fetch_recommended_apps_from_builtin(
+                    "en-US"
+                )
             )
 
-        categories = set()
-        recommended_apps_result = []
-        for recommended_app in recommended_apps:
-            app = recommended_app.app
-            if not app or not app.is_public:
-                continue
-
-            site = app.site
-            if not site:
-                continue
-
-            recommended_app_result = {
-                "id": recommended_app.id,
-                "app": {
-                    "id": app.id,
-                    "name": app.name,
-                    "mode": app.mode,
-                    "icon": app.icon,
-                    "icon_background": app.icon_background,
-                },
-                "app_id": recommended_app.app_id,
-                "description": site.description,
-                "copyright": site.copyright,
-                "privacy_policy": site.privacy_policy,
-                "custom_disclaimer": site.custom_disclaimer,
-                "category": recommended_app.category,
-                "position": recommended_app.position,
-                "is_listed": recommended_app.is_listed,
-            }
-            recommended_apps_result.append(recommended_app_result)
-
-            categories.add(recommended_app.category)  # add category to categories
-
-        return {"recommended_apps": recommended_apps_result, "categories": sorted(categories)}
-
-    @classmethod
-    def _fetch_recommended_apps_from_dify_official(cls, language: str) -> dict:
-        """
-        Fetch recommended apps from dify official.
-        :param language: language
-        :return:
-        """
-        domain = dify_config.HOSTED_FETCH_APP_TEMPLATES_REMOTE_DOMAIN
-        url = f"{domain}/apps?language={language}"
-        response = requests.get(url, timeout=(3, 10))
-        if response.status_code != 200:
-            raise ValueError(f"fetch recommended apps failed, status code: {response.status_code}")
-
-        result = response.json()
-
-        if "categories" in result:
-            result["categories"] = sorted(result["categories"])
-
         return result
 
-    @classmethod
-    def _fetch_recommended_apps_from_builtin(cls, language: str) -> dict:
-        """
-        Fetch recommended apps from builtin.
-        :param language: language
-        :return:
-        """
-        builtin_data = cls._get_builtin_data()
-        return builtin_data.get("recommended_apps", {}).get(language)
-
     @classmethod
     def get_recommend_app_detail(cls, app_id: str) -> Optional[dict]:
         """
         :return:
         """
         mode = dify_config.HOSTED_FETCH_APP_TEMPLATES_MODE
-        if mode == "remote":
-            try:
-                result = cls._fetch_recommended_app_detail_from_dify_official(app_id)
-            except Exception as e:
-                logger.warning(f"fetch recommended app detail from dify official failed: {e}, switch to built-in.")
-                result = cls._fetch_recommended_app_detail_from_builtin(app_id)
-        elif mode == "db":
-            result = cls._fetch_recommended_app_detail_from_db(app_id)
-        elif mode == "builtin":
-            result = cls._fetch_recommended_app_detail_from_builtin(app_id)
-        else:
-            raise ValueError(f"invalid fetch recommended app detail mode: {mode}")
-
+        retrieval_instance = RecommendAppRetrievalFactory.get_recommend_app_factory(mode)()
+        result = retrieval_instance.get_recommend_app_detail(app_id)
         return result
-
-    @classmethod
-    def _fetch_recommended_app_detail_from_dify_official(cls, app_id: str) -> Optional[dict]:
-        """
-        Fetch recommended app detail from dify official.
-        :param app_id: App ID
-        :return:
-        """
-        domain = dify_config.HOSTED_FETCH_APP_TEMPLATES_REMOTE_DOMAIN
-        url = f"{domain}/apps/{app_id}"
-        response = requests.get(url, timeout=(3, 10))
-        if response.status_code != 200:
-            return None
-
-        return response.json()
-
-    @classmethod
-    def _fetch_recommended_app_detail_from_db(cls, app_id: str) -> Optional[dict]:
-        """
-        Fetch recommended app detail from db.
-        :param app_id: App ID
-        :return:
-        """
-        # is in public recommended list
-        recommended_app = (
-            db.session.query(RecommendedApp)
-            .filter(RecommendedApp.is_listed == True, RecommendedApp.app_id == app_id)
-            .first()
-        )
-
-        if not recommended_app:
-            return None
-
-        # get app detail
-        app_model = db.session.query(App).filter(App.id == app_id).first()
-        if not app_model or not app_model.is_public:
-            return None
-
-        return {
-            "id": app_model.id,
-            "name": app_model.name,
-            "icon": app_model.icon,
-            "icon_background": app_model.icon_background,
-            "mode": app_model.mode,
-            "export_data": AppDslService.export_dsl(app_model=app_model),
-        }
-
-    @classmethod
-    def _fetch_recommended_app_detail_from_builtin(cls, app_id: str) -> Optional[dict]:
-        """
-        Fetch recommended app detail from builtin.
-        :param app_id: App ID
-        :return:
-        """
-        builtin_data = cls._get_builtin_data()
-        return builtin_data.get("app_details", {}).get(app_id)
-
-    @classmethod
-    def _get_builtin_data(cls) -> dict:
-        """
-        Get builtin data.
-        :return:
-        """
-        if cls.builtin_data:
-            return cls.builtin_data
-
-        root_path = current_app.root_path
-        cls.builtin_data = json.loads(
-            Path(path.join(root_path, "constants", "recommended_apps.json")).read_text(encoding="utf-8")
-        )
-
-        return cls.builtin_data
-
-    @classmethod
-    def fetch_all_recommended_apps_and_export_datas(cls):
-        """
-        Fetch all recommended apps and export datas
-        :return:
-        """
-        templates = {"recommended_apps": {}, "app_details": {}}
-        for language in languages:
-            try:
-                result = cls._fetch_recommended_apps_from_dify_official(language)
-            except Exception as e:
-                logger.warning(f"fetch recommended apps from dify official failed: {e}, skip.")
-                continue
-
-            templates["recommended_apps"][language] = result
-
-            for recommended_app in result.get("recommended_apps"):
-                app_id = recommended_app.get("app_id")
-
-                # get app detail
-                app_detail = cls._fetch_recommended_app_detail_from_dify_official(app_id)
-                if not app_detail:
-                    continue
-
-                templates["app_details"][app_id] = app_detail
-
-        return templates