Study/items title search

backend/app/api/routes/items.py

@@ -1,7 +1,7 @@
 import uuid
-from typing import Any
+from typing import Annotated, Any
 
-from fastapi import APIRouter, HTTPException
+from fastapi import APIRouter, HTTPException, Query
 from sqlmodel import col, func, select
 
 from app.api.deps import CurrentUser, SessionDep
@@ -12,34 +12,38 @@
 
 @router.get("/", response_model=ItemsPublic)
 def read_items(
-    session: SessionDep, current_user: CurrentUser, skip: int = 0, limit: int = 100
+    session: SessionDep,
+    current_user: CurrentUser,
+    skip: Annotated[int, Query(ge=0)] = 0,
+    limit: Annotated[int, Query(ge=1, le=100)] = 100,
+    q: str | None = None,
 ) -> Any:
     """
     Retrieve items.
     """
+    count_statement = select(func.count()).select_from(Item)
+    statement = select(Item)
+
+    # 普通用户只能看到自己的数据
+    if not current_user.is_superuser:
+        owner_condition = Item.owner_id == current_user.id
+        count_statement = count_statement.where(owner_condition)
+        statement = statement.where(owner_condition)
+
+    # 去除关键词两端的空格；空关键词不进行过滤
+    keyword = q.strip() if q else ""
+
+    if keyword:
+        search_condition = col(Item.title).icontains(keyword, autoescape=True)
+        count_statement = count_statement.where(search_condition)
+        statement = statement.where(search_condition)
+
+    count = session.exec(count_statement).one()
 
-    if current_user.is_superuser:
-        count_statement = select(func.count()).select_from(Item)
-        count = session.exec(count_statement).one()
-        statement = (
-            select(Item).order_by(col(Item.created_at).desc()).offset(skip).limit(limit)
-        )
-        items = session.exec(statement).all()
-    else:
-        count_statement = (
-            select(func.count())
-            .select_from(Item)
-            .where(Item.owner_id == current_user.id)
-        )
-        count = session.exec(count_statement).one()
-        statement = (
-            select(Item)
-            .where(Item.owner_id == current_user.id)
-            .order_by(col(Item.created_at).desc())
-            .offset(skip)
-            .limit(limit)
-        )
-        items = session.exec(statement).all()
+    statement = (
+        statement.order_by(col(Item.created_at).desc()).offset(skip).limit(limit)
+    )
+    items = session.exec(statement).all()
 
     items_public = [ItemPublic.model_validate(item) for item in items]
     return ItemsPublic(data=items_public, count=count)

backend/tests/api/routes/test_items.py

@@ -1,5 +1,6 @@
 import uuid
 
+import pytest
 from fastapi.testclient import TestClient
 from sqlmodel import Session
 
@@ -24,6 +25,38 @@ def test_create_item(
     assert "owner_id" in content
 
 
+def test_create_item_empty_title(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    data = {"title": "", "description": "测试空标题"}
+    response = client.post(
+        f"{settings.API_V1_STR}/items/",
+        headers=superuser_token_headers,
+        json=data,
+    )
+    # print(response.json())
+    assert response.status_code == 422
+
+
+def test_create_item_missing_title(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    data = {"description": "测试缺少标题"}
+    response = client.post(
+        f"{settings.API_V1_STR}/items/",
+        headers=superuser_token_headers,
+        json=data,
+    )
+    # print(response.json())
+    assert response.status_code == 422
+
+    errors = response.json()["detail"]
+    assert any(
+        error["loc"] == ["body", "title"] and error["type"] == "missing"
+        for error in errors
+    )
+
+
 def test_read_item(
     client: TestClient, superuser_token_headers: dict[str, str], db: Session
 ) -> None:
@@ -40,6 +73,262 @@ def test_read_item(
     assert content["owner_id"] == str(item.owner_id)
 
 
+def test_read_items_search_no_match(
+    client: TestClient,
+    superuser_token_headers: dict[str, str],
+) -> None:
+    created = client.post(
+        f"{settings.API_V1_STR}/items/",
+        headers=superuser_token_headers,
+        json={"title": "买牛奶"},
+    )
+    assert created.status_code == 200
+
+    response = client.get(
+        f"{settings.API_V1_STR}/items/",
+        headers=superuser_token_headers,
+        params={"q": uuid.uuid4().hex},
+    )
+
+    assert response.status_code == 200
+    content = response.json()
+    assert content["data"] == []
+    assert content["count"] == 0
+
+
+def test_read_items_search_ignores_case(
+    client: TestClient,
+    superuser_token_headers: dict[str, str],
+) -> None:
+    keyword = f"Hello{uuid.uuid4().hex}"
+
+    created = client.post(
+        f"{settings.API_V1_STR}/items/",
+        headers=superuser_token_headers,
+        json={"title": f"学习 {keyword.upper()}"},
+    )
+    assert created.status_code == 200
+    created_id = created.json()["id"]
+
+    response = client.get(
+        f"{settings.API_V1_STR}/items/",
+        headers=superuser_token_headers,
+        params={"q": keyword.lower()},
+    )
+
+    assert response.status_code == 200
+    content = response.json()
+    assert content["count"] == 1
+    assert [item["id"] for item in content["data"]] == [created_id]
+
+
+@pytest.mark.parametrize(
+    "search_params",
+    [
+        {},
+        {"q": ""},
+        {"q": "   "},
+    ],
+)
+def test_read_items_empty_search(
+    client: TestClient,
+    normal_user_token_headers: dict[str, str],
+    search_params: dict[str, str],
+) -> None:
+    expected_ids = set()
+
+    # 为当前普通用户准备两条数据
+    for title in ("学习 Python", "买牛奶"):
+        created = client.post(
+            f"{settings.API_V1_STR}/items/",
+            headers=normal_user_token_headers,
+            json={"title": title},
+        )
+        assert created.status_code == 200
+        expected_ids.add(created.json()["id"])
+
+    # 不搜索时的结果，作为比较基准
+    baseline = client.get(
+        f"{settings.API_V1_STR}/items/",
+        headers=normal_user_token_headers,
+        params={"skip": 0, "limit": 100},
+    )
+    assert baseline.status_code == 200
+    baseline_content = baseline.json()
+
+    response = client.get(
+        f"{settings.API_V1_STR}/items/",
+        headers=normal_user_token_headers,
+        params={"skip": 0, "limit": 100, **search_params},
+    )
+
+    assert response.status_code == 200
+    content = response.json()
+    assert content == baseline_content
+
+    baseline_ids = {item["id"] for item in baseline_content["data"]}
+    assert expected_ids <= baseline_ids
+
+
+def test_read_items_search_pagination_count(
+    client: TestClient,
+    superuser_token_headers: dict[str, str],
+) -> None:
+    keyword = uuid.uuid4().hex
+    matching_ids = set()
+
+    # 创建两条标题匹配的数据
+    for number in range(2):
+        created = client.post(
+            f"{settings.API_V1_STR}/items/",
+            headers=superuser_token_headers,
+            json={"title": f"{keyword} 第 {number + 1} 条"},
+        )
+        assert created.status_code == 200
+        matching_ids.add(created.json()["id"])
+
+    # 创建一条不匹配的数据，不应计入搜索总数
+    other = client.post(
+        f"{settings.API_V1_STR}/items/",
+        headers=superuser_token_headers,
+        json={"title": "买牛奶"},
+    )
+    assert other.status_code == 200
+
+    returned_ids = []
+
+    # 分别查询第一页和第二页
+    for skip in (0, 1):
+        response = client.get(
+            f"{settings.API_V1_STR}/items/",
+            headers=superuser_token_headers,
+            params={"q": keyword, "skip": skip, "limit": 1},
+        )
+
+        assert response.status_code == 200
+        content = response.json()
+        assert content["count"] == 2
+        assert len(content["data"]) == 1
+        returned_ids.append(content["data"][0]["id"])
+
+    assert set(returned_ids) == matching_ids
+
+
+def test_read_items_search_respects_owner(
+    client: TestClient,
+    normal_user_token_headers: dict[str, str],
+    superuser_token_headers: dict[str, str],
+) -> None:
+    keyword = uuid.uuid4().hex
+
+    # 普通用户创建自己的匹配数据
+    own = client.post(
+        f"{settings.API_V1_STR}/items/",
+        headers=normal_user_token_headers,
+        json={"title": f"自己的 {keyword}"},
+    )
+    assert own.status_code == 200
+    own_id = own.json()["id"]
+
+    # 管理员创建另一条同样匹配的数据
+    other = client.post(
+        f"{settings.API_V1_STR}/items/",
+        headers=superuser_token_headers,
+        json={"title": f"别人的 {keyword}"},
+    )
+    assert other.status_code == 200
+
+    # 普通用户搜索，只应看到自己的那条
+    response = client.get(
+        f"{settings.API_V1_STR}/items/",
+        headers=normal_user_token_headers,
+        params={"q": keyword},
+    )
+
+    assert response.status_code == 200
+    content = response.json()
+    assert content["count"] == 1
+    assert [item["id"] for item in content["data"]] == [own_id]
+
+
+def test_read_items_search_by_title(
+    client: TestClient,
+    superuser_token_headers: dict[str, str],
+) -> None:
+    keyword = uuid.uuid4().hex
+
+    # 创建标题匹配的数据
+    matching = client.post(
+        f"{settings.API_V1_STR}/items/",
+        headers=superuser_token_headers,
+        json={"title": f"学习 {keyword}"},
+    )
+    assert matching.status_code == 200
+    matching_id = matching.json()["id"]
+
+    # 创建标题不匹配、但描述包含关键词的数据
+    other = client.post(
+        f"{settings.API_V1_STR}/items/",
+        headers=superuser_token_headers,
+        json={"title": "买牛奶", "description": keyword},
+    )
+    assert other.status_code == 200
+
+    # 按标题关键词搜索
+    response = client.get(
+        f"{settings.API_V1_STR}/items/",
+        headers=superuser_token_headers,
+        params={"q": keyword},
+    )
+
+    assert response.status_code == 200
+    content = response.json()
+    assert content["count"] == 1
+    assert [item["id"] for item in content["data"]] == [matching_id]
+
+
+@pytest.mark.parametrize(
+    "params",
+    [
+        {"skip": -1},
+        {"limit": 0},
+        {"limit": 101},
+    ],
+)
+def test_read_items_invalid_pagination(
+    client: TestClient,
+    superuser_token_headers: dict[str, str],
+    params: dict[str, int],
+) -> None:
+    response = client.get(
+        f"{settings.API_V1_STR}/items/",
+        headers=superuser_token_headers,
+        params=params,
+    )
+    assert response.status_code == 422
+
+
+@pytest.mark.parametrize("limit", [1, 100])
+def test_read_items_valid_limit(
+    client: TestClient,
+    superuser_token_headers: dict[str, str],
+    db: Session,
+    limit: int,
+) -> None:
+    create_random_item(db)
+    create_random_item(db)
+
+    response = client.get(
+        f"{settings.API_V1_STR}/items/",
+        headers=superuser_token_headers,
+        params={"skip": 0, "limit": limit},
+    )
+
+    assert response.status_code == 200
+    content = response.json()
+    assert len(content["data"]) == min(limit, content["count"])
+
+
 def test_read_item_not_found(
     client: TestClient, superuser_token_headers: dict[str, str]
 ) -> None:

frontend/src/client/types.gen.ts

@@ -784,6 +784,10 @@ export type itemsReadItemsData = {
          * Limit
          */
         limit?: number;
+        /**
+         * Q
+         */
+        q?: string | null;
     };
     url: '/api/v1/items/';
 };