diff --git a/apps/api/app/api/routes/cover_templates.py b/apps/api/app/api/routes/cover_templates.py index 7f2c3b80b..f42cbfda2 100644 --- a/apps/api/app/api/routes/cover_templates.py +++ b/apps/api/app/api/routes/cover_templates.py @@ -20,6 +20,7 @@ from app.schemas.cover_template import ( UpdateCoverTemplateRequest, ) from fastapi import APIRouter, Depends, HTTPException, Response +from sqlalchemy.exc import OperationalError, ProgrammingError from packages.domain.cover_template import CoverTemplate @@ -35,10 +36,17 @@ def list_cover_templates( authenticated_user: AuthenticatedUser = Depends(get_current_user), repo: Any = Depends(get_cover_template_repository), ) -> ListCoverTemplatesResponse: - """列出当前用户可见的封面模板（系统模板 + 用户自定义模板）。""" + """列出当前用户可见的封面模板（系统模板 + 用户自定义模板）。 + + 当数据库表不存在时（迁移未执行），降级返回空列表而非 500。 + """ user_id = authenticated_user.user.id - items = repo.list_for_user(user_id, skip=skip, limit=limit) - total = repo.count_for_user(user_id) + try: + items = repo.list_for_user(user_id, skip=skip, limit=limit) + total = repo.count_for_user(user_id) + except (OperationalError, ProgrammingError) as exc: + logger.warning("cover_templates 表查询失败（可能未迁移），返回空列表: %s", exc) + return ListCoverTemplatesResponse(items=[], total=0) return ListCoverTemplatesResponse( items=[ CoverTemplateResponse( @@ -70,7 +78,11 @@ def create_cover_template( config=config_dict, thumbnail_url=request.thumbnail_url, ) - created = repo.create(template) + try: + created = repo.create(template) + except (OperationalError, ProgrammingError) as exc: + logger.warning("cover_templates 表不可用（可能未迁移）: %s", exc) + raise HTTPException(status_code=503, detail="封面模板服务暂不可用，请稍后重试") from None return CoverTemplateResponse( id=created.id, name=created.name, @@ -90,7 +102,11 @@ def update_cover_template( ) -> CoverTemplateResponse: """更新封面模板（仅允许更新自己的模板）。""" user_id = authenticated_user.user.id - template = repo.get(template_id) + try: + template = repo.get(template_id) + except (OperationalError, ProgrammingError) as exc: + logger.warning("cover_templates 表不可用: %s", exc) + raise HTTPException(status_code=503, detail="封面模板服务暂不可用，请稍后重试") from None if template is None: raise HTTPException(status_code=404, detail="模板不存在") if template.is_system: @@ -124,7 +140,11 @@ def delete_cover_template( ) -> None: """删除用户自定义封面模板（系统模板不可删除）。""" user_id = authenticated_user.user.id - template = repo.get(template_id) + try: + template = repo.get(template_id) + except (OperationalError, ProgrammingError) as exc: + logger.warning("cover_templates 表不可用: %s", exc) + raise HTTPException(status_code=503, detail="封面模板服务暂不可用，请稍后重试") from None if template is None: raise HTTPException(status_code=404, detail="模板不存在") if template.is_system: diff --git a/tests/unit/test_cover_templates_crud.py b/tests/unit/test_cover_templates_crud.py index c6a61c80c..9ef2f88a1 100644 --- a/tests/unit/test_cover_templates_crud.py +++ b/tests/unit/test_cover_templates_crud.py @@ -307,3 +307,96 @@ class TestInMemoryCoverTemplateRepository: system = repo.list_system_templates() assert len(system) == 2 assert all(t.is_system for t in system) + + +class TestCoverTemplatesErrorHandling: + """测试 API 错误处理加固 - 数据库表不存在时降级处理。""" + + def test_list_raises_operational_error_when_table_missing(self): + """当 cover_templates 表不存在时，repository 应抛出 OperationalError。""" + from sqlalchemy.exc import OperationalError + + from packages.adapters.sqlalchemy_impl.cover_template_repository import SQLAlchemyCoverTemplateRepository + + mock_session = MagicMock() + mock_session.query.side_effect = OperationalError( + "statement", {}, Exception('relation "cover_templates" does not exist') + ) + repo = SQLAlchemyCoverTemplateRepository(mock_session) + + with pytest.raises(OperationalError): + repo.list_for_user("user-1") + + def test_route_catches_operational_error_on_list(self): + """路由层应捕获 OperationalError 并返回空列表。""" + from app.api.routes.cover_templates import list_cover_templates + from app.schemas.cover_template import ListCoverTemplatesResponse + from sqlalchemy.exc import OperationalError + + mock_repo = MagicMock() + mock_repo.list_for_user.side_effect = OperationalError( + "statement", {}, Exception('relation "cover_templates" does not exist') + ) + + mock_user = MagicMock() + mock_user.user.id = "user-1" + + # 调用路由函数，验证它不会抛出异常 + result = list_cover_templates( + skip=0, + limit=100, + authenticated_user=mock_user, + repo=mock_repo, + ) + assert isinstance(result, ListCoverTemplatesResponse) + assert result.items == [] + assert result.total == 0 + + def test_route_catches_programming_error_on_list(self): + """路由层应捕获 ProgrammingError 并返回空列表。""" + from app.api.routes.cover_templates import list_cover_templates + from app.schemas.cover_template import ListCoverTemplatesResponse + from sqlalchemy.exc import ProgrammingError + + mock_repo = MagicMock() + mock_repo.list_for_user.side_effect = ProgrammingError( + "statement", {}, Exception("no such table: cover_templates") + ) + + mock_user = MagicMock() + mock_user.user.id = "user-1" + + result = list_cover_templates( + skip=0, + limit=100, + authenticated_user=mock_user, + repo=mock_repo, + ) + assert isinstance(result, ListCoverTemplatesResponse) + assert result.items == [] + assert result.total == 0 + + def test_create_returns_503_when_table_missing(self): + """创建模板时，如果表不存在应抛出 HTTPException(503)。""" + from app.api.routes.cover_templates import create_cover_template + from app.schemas.cover_template import CreateCoverTemplateRequest + from fastapi import HTTPException + from sqlalchemy.exc import OperationalError + + mock_repo = MagicMock() + mock_repo.create.side_effect = OperationalError( + "statement", {}, Exception('relation "cover_templates" does not exist') + ) + + mock_user = MagicMock() + mock_user.user.id = "user-1" + + request = CreateCoverTemplateRequest(name="test") + + with pytest.raises(HTTPException) as exc_info: + create_cover_template( + request=request, + authenticated_user=mock_user, + repo=mock_repo, + ) + assert exc_info.value.status_code == 503