diff --git a/apps/manager/appcenter.py b/apps/manager/appcenter.py index 6f80c37dd7ceeab7990d53d7d66f863693d693c6..d6d398a4abb067a7beb5bbc07403ba4f2a0c5d8d 100644 --- a/apps/manager/appcenter.py +++ b/apps/manager/appcenter.py @@ -2,6 +2,7 @@ Copyright (c) Huawei Technologies Co., Ltd. 2024-2025. All rights reserved. """ + import uuid from datetime import datetime, timezone from enum import Enum @@ -50,11 +51,15 @@ class AppCenterManager: try: # 搜索条件，仅显示已发布的应用 base_filter = {"published": True} - filters: dict[str, Any] = AppCenterManager._build_filters( - {"published": True}, - search_type, - keyword, - ) if keyword else base_filter + filters: dict[str, Any] = ( + AppCenterManager._build_filters( + {"published": True}, + search_type, + keyword, + ) + if keyword + else base_filter + ) # 执行应用搜索 apps, total_apps = await AppCenterManager._search_apps_by_filter(filters, page, page_size) # 构建返回的应用卡片列表 @@ -95,12 +100,16 @@ class AppCenterManager: try: # 搜索条件 base_filter = {"author": user_sub} - filters: dict[str, Any] = AppCenterManager._build_filters( - base_filter, - search_type, - keyword, - ) if keyword and search_type != SearchType.AUTHOR else base_filter - apps, total_apps= await AppCenterManager._search_apps_by_filter(filters, page, page_size) + filters: dict[str, Any] = ( + AppCenterManager._build_filters( + base_filter, + search_type, + keyword, + ) + if keyword and search_type != SearchType.AUTHOR + else base_filter + ) + apps, total_apps = await AppCenterManager._search_apps_by_filter(filters, page, page_size) return [ AppCenterCardItem( appId=app.id, @@ -141,11 +150,15 @@ class AppCenterManager: "_id": {"$in": fav_app}, "published": True, } - filters: dict[str, Any] = AppCenterManager._build_filters( - base_filter, - search_type, - keyword, - ) if keyword else base_filter + filters: dict[str, Any] = ( + AppCenterManager._build_filters( + base_filter, + search_type, + keyword, + ) + if keyword + else base_filter + ) apps, total_apps = await AppCenterManager._search_apps_by_filter(filters, page, page_size) return [ AppCenterCardItem( @@ -306,6 +319,30 @@ class AppCenterManager: LOGGER.error(f"[AppCenterManager] Modify favorite app failed: {e}") return AppCenterManager.ModFavAppFlag.INTERNAL_ERROR + @staticmethod + async def delete_app(app_id: str, user_sub: str) -> bool: + """删除应用 + + :param app_id: 应用唯一标识 + :param user_sub: 用户唯一标识 + :return: 删除是否成功 + """ + try: + async with MongoDB.get_session() as session, await session.start_transaction(): + app_collection = MongoDB.get_collection("app") + await app_collection.delete_one({"_id": app_id}, session=session) + user_collection = MongoDB.get_collection("user") + await user_collection.update_one( + {"_id": user_sub}, + {"$unset": {f"app_usage.{app_id}": ""}}, + session=session, + ) + await session.commit_transaction() + return True + except Exception as e: + LOGGER.error(f"[AppCenterManager] Delete app failed: {e}") + return False + @staticmethod async def get_recently_used_apps(count: int, user_sub: str) -> Optional[RecentAppList]: """获取用户最近使用的应用列表 @@ -331,40 +368,15 @@ class AppCenterManager: apps = [] # 如果 app_ids 为空，直接返回空列表 else: # 查询 MongoDB，获取符合条件的应用 - apps = await app_collection.find({"_id": {"$in": app_ids}}, {"name": 1}).to_list(len(app_ids)) # 传入 app_ids 的长度 + apps = await app_collection.find({"_id": {"$in": app_ids}}, {"name": 1}).to_list(len(app_ids)) app_map = {str(a["_id"]): a.get("name", "") for a in apps} - return RecentAppList(applications=[ - RecentAppListItem(appId=app_id, name=app_map.get(app_id, "")) - for app_id in app_ids - ]) + return RecentAppList( + applications=[RecentAppListItem(appId=app_id, name=app_map.get(app_id, "")) for app_id in app_ids], + ) except Exception as e: LOGGER.info(f"[AppCenterManager] Get recently used apps failed: {e}") return None - @staticmethod - async def delete_app(app_id: str, user_sub: str) -> bool: - """删除应用 - - :param app_id: 应用唯一标识 - :param user_sub: 用户唯一标识 - :return: 删除是否成功 - """ - try: - async with MongoDB.get_session() as session, await session.start_transaction(): - app_collection = MongoDB.get_collection("app") - await app_collection.delete_one({"_id": app_id}, session=session) - user_collection = MongoDB.get_collection("user") - await user_collection.update_one( - {"_id": user_sub}, - {"$unset": {f"app_usage.{app_id}": ""}}, - session=session, - ) - await session.commit_transaction() - return True - except Exception as e: - LOGGER.error(f"[AppCenterManager] Delete app failed: {e}") - return False - @staticmethod async def update_recent_app(user_sub: str, app_id: str) -> bool: """更新用户的最近使用应用列表 @@ -428,11 +440,13 @@ class AppCenterManager: try: app_collection = MongoDB.get_collection("app") total_apps = await app_collection.count_documents(search_conditions) - db_data = await app_collection.find(search_conditions) \ - .sort("created_at", -1) \ - .skip((page - 1) * page_size) \ - .limit(page_size) \ + db_data = ( + await app_collection.find(search_conditions) + .sort("created_at", -1) + .skip((page - 1) * page_size) + .limit(page_size) .to_list(length=page_size) + ) apps = [AppPool.model_validate(doc) for doc in db_data] return apps, total_apps except Exception as e: diff --git a/apps/routers/appcenter.py b/apps/routers/appcenter.py index 630daf03c72412fda5b9f36b80a62c3309e1d669..990f478d6e0b5769c765bee222261c068b69fc59 100644 --- a/apps/routers/appcenter.py +++ b/apps/routers/appcenter.py @@ -2,6 +2,7 @@ Copyright (c) Huawei Technologies Co., Ltd. 2024-2025. All rights reserved. """ + from typing import Annotated, Optional, Union from fastapi import APIRouter, Body, Depends, Path, Query, status @@ -35,7 +36,7 @@ router = APIRouter( @router.get("", response_model=Union[GetAppListRsp, ResponseData]) -async def get_applications( # noqa: ANN201, PLR0913 +async def get_applications( # noqa: PLR0913 user_sub: Annotated[str, Depends(get_user)], *, my_app: Annotated[bool, Query(..., alias="createdByMe", description="筛选我创建的")] = False, @@ -43,132 +44,167 @@ async def get_applications( # noqa: ANN201, PLR0913 search_type: Annotated[SearchType, Query(..., alias="searchType", description="搜索类型")] = SearchType.ALL, keyword: Annotated[Optional[str], Query(..., alias="keyword", description="搜索关键字")] = None, page: Annotated[int, Query(..., alias="page", ge=1, description="页码")] = 1, - page_size: Annotated[int, Query(..., alias="pageSize", ge=1, le=100, description="每页条数")] = 20, -): + page_size: Annotated[int, Query(..., alias="pageSize", ge=1, le=100, description="每页条数")] = 16, +) -> JSONResponse: """获取应用列表""" if my_app and my_fav: # 只能同时使用一个过滤条件 - return ResponseData( - code=status.HTTP_400_BAD_REQUEST, - message="createdByMe 和 favorited 不能同时生效", - result={}, + return JSONResponse( + status_code=status.HTTP_400_BAD_REQUEST, + content=ResponseData( + code=status.HTTP_400_BAD_REQUEST, + message="createdByMe 和 favorited 不能同时生效", + result={}, + ).model_dump(exclude_none=True, by_alias=True), ) app_cards, total_apps = [], -1 - if my_app: # 筛选我创建的 - app_cards, total_apps = await AppCenterManager.fetch_user_apps( - user_sub, search_type, keyword, page, page_size) + if my_app: # 筛选我创建的 + app_cards, total_apps = await AppCenterManager.fetch_user_apps(user_sub, search_type, keyword, page, page_size) elif my_fav: # 筛选已收藏的 app_cards, total_apps = await AppCenterManager.fetch_favorite_apps( - user_sub, search_type, keyword, page, page_size) - else: # 获取所有应用 - app_cards, total_apps = await AppCenterManager.fetch_all_apps( - user_sub, search_type, keyword, page, page_size) + user_sub, + search_type, + keyword, + page, + page_size, + ) + else: # 获取所有应用 + app_cards, total_apps = await AppCenterManager.fetch_all_apps(user_sub, search_type, keyword, page, page_size) if total_apps == -1: - return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content=ResponseData( - code=status.HTTP_500_INTERNAL_SERVER_ERROR, - message="查询失败", - result={}, - ).model_dump(exclude_none=True, by_alias=True)) - return JSONResponse(status_code=status.HTTP_200_OK, content=GetAppListRsp( - code=status.HTTP_200_OK, - message="查询成功", - result=GetAppListMsg( - currentPage=page, - totalApps=total_apps, - applications=app_cards, - ), - ).model_dump(exclude_none=True, by_alias=True)) + return JSONResponse( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + content=ResponseData( + code=status.HTTP_500_INTERNAL_SERVER_ERROR, + message="查询失败", + result={}, + ).model_dump(exclude_none=True, by_alias=True), + ) + return JSONResponse( + status_code=status.HTTP_200_OK, + content=GetAppListRsp( + code=status.HTTP_200_OK, + message="查询成功", + result=GetAppListMsg( + currentPage=page, + totalApps=total_apps, + applications=app_cards, + ), + ).model_dump(exclude_none=True, by_alias=True), + ) @router.post("", dependencies=[Depends(verify_csrf_token)], response_model=Union[BaseAppOperationRsp, ResponseData]) -async def create_or_update_application( # noqa: ANN201 +async def create_or_update_application( request: Annotated[CreateAppRequest, Body(...)], user_sub: Annotated[str, Depends(get_user)], -): +) -> JSONResponse: """创建或更新应用""" app_id = request.app_id if app_id: # 更新应用 confirm = await AppCenterManager.update_app(app_id, request) if not confirm: - return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content=ResponseData( + return JSONResponse( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + content=ResponseData( + code=status.HTTP_500_INTERNAL_SERVER_ERROR, + message="更新失败", + result={}, + ).model_dump(exclude_none=True, by_alias=True), + ) + return JSONResponse( + status_code=status.HTTP_200_OK, + content=BaseAppOperationRsp( + code=status.HTTP_200_OK, + message="更新成功", + result=BaseAppOperationMsg(appId=app_id), + ).model_dump(exclude_none=True, by_alias=True), + ) + # 创建应用 + app_id = await AppCenterManager.create_app(user_sub, request) + if not app_id: + return JSONResponse( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + content=ResponseData( code=status.HTTP_500_INTERNAL_SERVER_ERROR, - message="更新失败", + message="创建失败", result={}, - ).model_dump(exclude_none=True, by_alias=True)) - return JSONResponse(status_code=status.HTTP_200_OK, content=BaseAppOperationRsp( + ).model_dump(exclude_none=True, by_alias=True), + ) + return JSONResponse( + status_code=status.HTTP_200_OK, + content=BaseAppOperationRsp( code=status.HTTP_200_OK, - message="更新成功", + message="创建成功", result=BaseAppOperationMsg(appId=app_id), - ).model_dump(exclude_none=True, by_alias=True)) - # 创建应用 - app_id = await AppCenterManager.create_app(user_sub, request) - if not app_id: - return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content=ResponseData( - code=status.HTTP_500_INTERNAL_SERVER_ERROR, - message="创建失败", - result={}, - ).model_dump(exclude_none=True, by_alias=True)) - return JSONResponse(status_code=status.HTTP_200_OK, content=BaseAppOperationRsp( - code=status.HTTP_200_OK, - message="创建成功", - result=BaseAppOperationMsg(appId=app_id), - ).model_dump(exclude_none=True, by_alias=True)) + ).model_dump(exclude_none=True, by_alias=True), + ) @router.get("/recent", response_model=Union[GetRecentAppListRsp, ResponseData]) -async def get_recently_used_applications( # noqa: ANN201 +async def get_recently_used_applications( user_sub: Annotated[str, Depends(get_user)], count: Annotated[int, Query(..., ge=1, le=10)] = 5, -): +) -> JSONResponse: """获取最近使用的应用""" - recent_apps = await AppCenterManager.get_recently_used_apps( - count, user_sub) + recent_apps = await AppCenterManager.get_recently_used_apps(count, user_sub) if recent_apps is None: - return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content=ResponseData( - code=status.HTTP_500_INTERNAL_SERVER_ERROR, - message="查询失败", - result={}, - ).model_dump(exclude_none=True, by_alias=True)) - return JSONResponse(status_code=status.HTTP_200_OK, content=GetRecentAppListRsp( - code=status.HTTP_200_OK, - message="查询成功", - result=recent_apps, - ).model_dump(exclude_none=True, by_alias=True)) + return JSONResponse( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + content=ResponseData( + code=status.HTTP_500_INTERNAL_SERVER_ERROR, + message="查询失败", + result={}, + ).model_dump(exclude_none=True, by_alias=True), + ) + return JSONResponse( + status_code=status.HTTP_200_OK, + content=GetRecentAppListRsp( + code=status.HTTP_200_OK, + message="查询成功", + result=recent_apps, + ).model_dump(exclude_none=True, by_alias=True), + ) @router.get("/{appId}", response_model=Union[GetAppPropertyRsp, ResponseData]) -async def get_application( # noqa: ANN201 +async def get_application( app_id: Annotated[str, Path(..., alias="appId", description="应用ID")], -): +) -> JSONResponse: """获取应用详情""" app_data = await AppCenterManager.fetch_app_data_by_id(app_id) if not app_data: - return JSONResponse(status_code=status.HTTP_404_NOT_FOUND, content=ResponseData( - code=status.HTTP_404_NOT_FOUND, - message="找不到应用", - result={}, - ).model_dump(exclude_none=True, by_alias=True)) - workflows = [{"id":flow.id,"name":flow.name} for flow in app_data.flows] - return JSONResponse(status_code=status.HTTP_200_OK, content=GetAppPropertyRsp( - code=status.HTTP_200_OK, - message="查询成功", - result=GetAppPropertyMsg( - appId=app_data.id, - published=app_data.published, - name=app_data.name, - description=app_data.description, - icon=app_data.icon, - links=app_data.links, - recommendedQuestions=app_data.first_questions, - dialogRounds=app_data.history_len, - permission=AppPermissionData( - visibility=app_data.permission.type, - authorizedUsers=app_data.permission.users, + return JSONResponse( + status_code=status.HTTP_404_NOT_FOUND, + content=ResponseData( + code=status.HTTP_404_NOT_FOUND, + message="找不到应用", + result={}, + ).model_dump(exclude_none=True, by_alias=True), + ) + workflows = [{"id": flow.id, "name": flow.name} for flow in app_data.flows] + return JSONResponse( + status_code=status.HTTP_200_OK, + content=GetAppPropertyRsp( + code=status.HTTP_200_OK, + message="查询成功", + result=GetAppPropertyMsg( + appId=app_data.id, + published=app_data.published, + name=app_data.name, + description=app_data.description, + icon=app_data.icon, + links=app_data.links, + recommendedQuestions=app_data.first_questions, + dialogRounds=app_data.history_len, + permission=AppPermissionData( + visibility=app_data.permission.type, + authorizedUsers=app_data.permission.users, + ), + workflows=workflows, ), - workflows=workflows, - ), - ).model_dump(exclude_none=True, by_alias=True)) + ).model_dump(exclude_none=True, by_alias=True), + ) @router.delete( @@ -176,112 +212,151 @@ async def get_application( # noqa: ANN201 dependencies=[Depends(verify_csrf_token)], response_model=Union[BaseAppOperationRsp, ResponseData], ) -async def delete_application( # noqa: ANN201 +async def delete_application( app_id: Annotated[str, Path(..., alias="appId", description="应用ID")], user_sub: Annotated[str, Depends(get_user)], -): +) -> JSONResponse: """删除应用""" app_data = await AppCenterManager.fetch_app_data_by_id(app_id) if not app_data: - return JSONResponse(status_code=status.HTTP_400_BAD_REQUEST, content=ResponseData( - code=status.HTTP_400_BAD_REQUEST, - message="查询失败", - result={}, - ).model_dump(exclude_none=True, by_alias=True)) + return JSONResponse( + status_code=status.HTTP_400_BAD_REQUEST, + content=ResponseData( + code=status.HTTP_400_BAD_REQUEST, + message="查询失败", + result={}, + ).model_dump(exclude_none=True, by_alias=True), + ) # 校验应用作者是否为当前用户 if app_data.author != user_sub: - return JSONResponse(status_code=status.HTTP_403_FORBIDDEN, content=ResponseData( - code=status.HTTP_403_FORBIDDEN, - message="无权删除他人创建的应用", - result={}, - ).model_dump(exclude_none=True, by_alias=True)) + return JSONResponse( + status_code=status.HTTP_403_FORBIDDEN, + content=ResponseData( + code=status.HTTP_403_FORBIDDEN, + message="无权删除他人创建的应用", + result={}, + ).model_dump(exclude_none=True, by_alias=True), + ) # 删除应用相关的工作流 for flow in app_data.flows: if not await FlowManager.delete_flow_by_app_and_flow_id(app_id, flow.id): - return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content=ResponseData( - code=status.HTTP_500_INTERNAL_SERVER_ERROR, - message=f"删除应用下属工作流 {flow.id} 失败", - result={}, - ).model_dump(exclude_none=True, by_alias=True)) + return JSONResponse( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + content=ResponseData( + code=status.HTTP_500_INTERNAL_SERVER_ERROR, + message=f"删除应用下属工作流 {flow.id} 失败", + result={}, + ).model_dump(exclude_none=True, by_alias=True), + ) # 删除应用 if not await AppCenterManager.delete_app(app_id, user_sub): - return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content=ResponseData( - code=status.HTTP_500_INTERNAL_SERVER_ERROR, - message="删除失败", - result={}, - ).model_dump(exclude_none=True, by_alias=True)) - return JSONResponse(status_code=status.HTTP_200_OK, content=BaseAppOperationRsp( - code=status.HTTP_200_OK, - message="删除成功", - result=BaseAppOperationMsg(appId=app_id), - ).model_dump(exclude_none=True, by_alias=True)) + return JSONResponse( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + content=ResponseData( + code=status.HTTP_500_INTERNAL_SERVER_ERROR, + message="删除失败", + result={}, + ).model_dump(exclude_none=True, by_alias=True), + ) + return JSONResponse( + status_code=status.HTTP_200_OK, + content=BaseAppOperationRsp( + code=status.HTTP_200_OK, + message="删除成功", + result=BaseAppOperationMsg(appId=app_id), + ).model_dump(exclude_none=True, by_alias=True), + ) @router.post("/{appId}", dependencies=[Depends(verify_csrf_token)], response_model=BaseAppOperationRsp) -async def publish_application( # noqa: ANN201 +async def publish_application( app_id: Annotated[str, Path(..., alias="appId", description="应用ID")], user_sub: Annotated[str, Depends(get_user)], -): +) -> JSONResponse: """发布应用""" app_data = await AppCenterManager.fetch_app_data_by_id(app_id) if not app_data: - return JSONResponse(status_code=status.HTTP_400_BAD_REQUEST, content=ResponseData( - code=status.HTTP_400_BAD_REQUEST, - message="查询失败", - result={}, - ).model_dump(exclude_none=True, by_alias=True)) + return JSONResponse( + status_code=status.HTTP_400_BAD_REQUEST, + content=ResponseData( + code=status.HTTP_400_BAD_REQUEST, + message="查询失败", + result={}, + ).model_dump(exclude_none=True, by_alias=True), + ) # 验证用户权限 if app_data.author != user_sub: - return JSONResponse(status_code=status.HTTP_403_FORBIDDEN, content=ResponseData( - code=status.HTTP_403_FORBIDDEN, - message="无权发布他人创建的应用", - result={}, - ).model_dump(exclude_none=True, by_alias=True)) + return JSONResponse( + status_code=status.HTTP_403_FORBIDDEN, + content=ResponseData( + code=status.HTTP_403_FORBIDDEN, + message="无权发布他人创建的应用", + result={}, + ).model_dump(exclude_none=True, by_alias=True), + ) # 发布应用 if not await AppCenterManager.publish_app(app_id): - return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content=ResponseData( - code=status.HTTP_500_INTERNAL_SERVER_ERROR, - message="发布失败", - result={}, - ).model_dump(exclude_none=True, by_alias=True)) - return JSONResponse(status_code=status.HTTP_200_OK, content=BaseAppOperationRsp( - code=status.HTTP_200_OK, - message="发布成功", - result=BaseAppOperationMsg(appId=app_id), - ).model_dump(exclude_none=True, by_alias=True)) + return JSONResponse( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + content=ResponseData( + code=status.HTTP_500_INTERNAL_SERVER_ERROR, + message="发布失败", + result={}, + ).model_dump(exclude_none=True, by_alias=True), + ) + return JSONResponse( + status_code=status.HTTP_200_OK, + content=BaseAppOperationRsp( + code=status.HTTP_200_OK, + message="发布成功", + result=BaseAppOperationMsg(appId=app_id), + ).model_dump(exclude_none=True, by_alias=True), + ) @router.put("/{appId}", dependencies=[Depends(verify_csrf_token)], response_model=Union[ModFavAppRsp, ResponseData]) -async def modify_favorite_application( # noqa: ANN201 +async def modify_favorite_application( app_id: Annotated[str, Path(..., alias="appId", description="应用ID")], request: Annotated[ModFavAppRequest, Body(...)], user_sub: Annotated[str, Depends(get_user)], -): +) -> JSONResponse: """更改应用收藏状态""" flag = await AppCenterManager.modify_favorite_app(app_id, user_sub, favorited=request.favorited) if flag == AppCenterManager.ModFavAppFlag.NOT_FOUND: - return JSONResponse(status_code=status.HTTP_404_NOT_FOUND, content=ResponseData( - code=status.HTTP_404_NOT_FOUND, - message="找不到应用", - result={}, - ).model_dump(exclude_none=True, by_alias=True)) + return JSONResponse( + status_code=status.HTTP_404_NOT_FOUND, + content=ResponseData( + code=status.HTTP_404_NOT_FOUND, + message="找不到应用", + result={}, + ).model_dump(exclude_none=True, by_alias=True), + ) if flag == AppCenterManager.ModFavAppFlag.BAD_REQUEST: - return JSONResponse(status_code=status.HTTP_400_BAD_REQUEST, content=ResponseData( + return JSONResponse( + status_code=status.HTTP_400_BAD_REQUEST, + content=ResponseData( code=status.HTTP_400_BAD_REQUEST, message="不可重复操作", result={}, - ).model_dump(exclude_none=True, by_alias=True)) + ).model_dump(exclude_none=True, by_alias=True), + ) if flag == AppCenterManager.ModFavAppFlag.INTERNAL_ERROR: - return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content=ResponseData( - code=status.HTTP_500_INTERNAL_SERVER_ERROR, - message="操作失败", - result={}, - ).model_dump(exclude_none=True, by_alias=True)) - return JSONResponse(status_code=status.HTTP_200_OK, content=ModFavAppRsp( - code=status.HTTP_200_OK, - message="操作成功", - result=ModFavAppMsg( - appId=app_id, - favorited=request.favorited, - ), - ).model_dump(exclude_none=True, by_alias=True)) + return JSONResponse( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + content=ResponseData( + code=status.HTTP_500_INTERNAL_SERVER_ERROR, + message="操作失败", + result={}, + ).model_dump(exclude_none=True, by_alias=True), + ) + return JSONResponse( + status_code=status.HTTP_200_OK, + content=ModFavAppRsp( + code=status.HTTP_200_OK, + message="操作成功", + result=ModFavAppMsg( + appId=app_id, + favorited=request.favorited, + ), + ).model_dump(exclude_none=True, by_alias=True), + )