Implement global search

backend/app/api/main.py

@@ -1,9 +1,10 @@
 from fastapi import APIRouter
 
-from app.api.routes import items, login, users, utils
+from app.api.routes import items, login, search, users, utils
 
 api_router = APIRouter()
 api_router.include_router(login.router, tags=["login"])
 api_router.include_router(users.router, prefix="/users", tags=["users"])
 api_router.include_router(utils.router, prefix="/utils", tags=["utils"])
 api_router.include_router(items.router, prefix="/items", tags=["items"])
+api_router.include_router(search.router, prefix="/search", tags=["search"])

backend/app/api/routes/search.py

@@ -0,0 +1,84 @@
+from typing import Any
+
+from fastapi import APIRouter, Query
+from sqlalchemy import or_
+from sqlmodel import col, func, select
+
+from app.api.deps import CurrentUser, SessionDep
+from app.models import Item, SearchResult, SearchResultsPublic, User
+
+router = APIRouter()
+
+
+@router.get("/", response_model=SearchResultsPublic)
+def search(
+    session: SessionDep,
+    current_user: CurrentUser,
+    q: str = Query(..., min_length=1),
+    limit: int = Query(default=10, ge=1, le=50),
+) -> Any:
+    """
+    Search across resources visible to the current user.
+    """
+    query = q.strip()
+    if not query:
+        return SearchResultsPublic(data=[], count=0)
+
+    pattern = f"%{query}%"
+    item_filters = or_(
+        col(Item.title).ilike(pattern),
+        col(Item.description).ilike(pattern),
+    )
+    item_statement = select(Item).where(item_filters)
+    item_count_statement = select(func.count()).select_from(Item).where(item_filters)
+
+    if not current_user.is_superuser:
+        item_statement = item_statement.where(Item.owner_id == current_user.id)
+        item_count_statement = item_count_statement.where(
+            Item.owner_id == current_user.id
+        )
+
+    total_count = session.exec(item_count_statement).one()
+    items = session.exec(
+        item_statement.order_by(col(Item.id).desc()).limit(limit)
+    ).all()
+    results = [
+        SearchResult(
+            type="item",
+            id=item.id,
+            title=item.title,
+            description=item.description,
+        )
+        for item in items
+        if item.id is not None
+    ]
+
+    if current_user.is_superuser:
+        user_filters = or_(
+            col(User.email).ilike(pattern),
+            col(User.full_name).ilike(pattern),
+        )
+        total_count += session.exec(
+            select(func.count()).select_from(User).where(user_filters)
+        ).one()
+
+        remaining_limit = limit - len(results)
+        if remaining_limit > 0:
+            users = session.exec(
+                select(User)
+                .where(user_filters)
+                .order_by(col(User.id).desc())
+                .limit(remaining_limit)
+            ).all()
+            results.extend(
+                SearchResult(
+                    type="user",
+                    id=user.id,
+                    title=user.full_name or user.email,
+                    description=user.email,
+                )
+                for user in users
+                if user.id is not None
+            )
+
+    return SearchResultsPublic(data=results, count=total_count)

backend/app/models.py

@@ -1,3 +1,5 @@
+from typing import Literal
+
 from sqlmodel import Field, Relationship, SQLModel
 
 
@@ -92,6 +94,18 @@ class ItemsPublic(SQLModel):
     count: int
 
 
+class SearchResult(SQLModel):
+    type: Literal["item", "user"]
+    id: int
+    title: str
+    description: str | None = None
+
+
+class SearchResultsPublic(SQLModel):
+    data: list[SearchResult]
+    count: int
+
+
 # Generic message
 class Message(SQLModel):
     message: str

backend/app/tests/api/routes/test_search.py

@@ -0,0 +1,143 @@
+from fastapi.testclient import TestClient
+from sqlmodel import Session
+
+from app import crud
+from app.core.config import settings
+from app.models import ItemCreate, UserCreate
+from app.tests.utils.utils import random_email, random_lower_string
+
+
+def test_search_returns_items_and_users_for_superuser(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    needle = f"search-{random_lower_string()}"
+    user = crud.create_user(
+        session=db,
+        user_create=UserCreate(
+            email=random_email(),
+            password="password",
+            full_name=f"{needle} user",
+        ),
+    )
+    assert user.id is not None
+    item = crud.create_item(
+        session=db,
+        item_in=ItemCreate(
+            title=f"{needle} item",
+            description="Visible to superusers",
+        ),
+        owner_id=user.id,
+    )
+
+    response = client.get(
+        f"{settings.API_V1_STR}/search/",
+        headers=superuser_token_headers,
+        params={"q": needle},
+    )
+
+    assert response.status_code == 200
+    content = response.json()
+    assert content["count"] >= 2
+    assert {
+        "type": "item",
+        "id": item.id,
+        "title": item.title,
+        "description": item.description,
+    } in content["data"]
+    assert {
+        "type": "user",
+        "id": user.id,
+        "title": user.full_name,
+        "description": user.email,
+    } in content["data"]
+
+
+def test_search_limits_normal_user_to_own_items(
+    client: TestClient,
+    normal_user_token_headers: dict[str, str],
+    db: Session,
+) -> None:
+    needle = f"private-{random_lower_string()}"
+    normal_user = crud.get_user_by_email(session=db, email=settings.EMAIL_TEST_USER)
+    assert normal_user is not None
+    assert normal_user.id is not None
+
+    own_item = crud.create_item(
+        session=db,
+        item_in=ItemCreate(title=f"{needle} visible", description="Owned match"),
+        owner_id=normal_user.id,
+    )
+    other_user = crud.create_user(
+        session=db,
+        user_create=UserCreate(email=random_email(), password="password"),
+    )
+    assert other_user.id is not None
+    hidden_item = crud.create_item(
+        session=db,
+        item_in=ItemCreate(title=f"{needle} hidden", description="Hidden match"),
+        owner_id=other_user.id,
+    )
+
+    response = client.get(
+        f"{settings.API_V1_STR}/search/",
+        headers=normal_user_token_headers,
+        params={"q": needle},
+    )
+
+    assert response.status_code == 200
+    content = response.json()
+    result_ids = {
+        (result["type"], result["id"]) for result in content["data"]
+    }
+    assert ("item", own_item.id) in result_ids
+    assert ("item", hidden_item.id) not in result_ids
+    assert all(result["type"] == "item" for result in content["data"])
+
+
+def test_search_count_includes_matches_beyond_limit(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    needle = f"limit-{random_lower_string()}"
+    user = crud.create_user(
+        session=db,
+        user_create=UserCreate(email=random_email(), password="password"),
+    )
+    assert user.id is not None
+    items = [
+        crud.create_item(
+            session=db,
+            item_in=ItemCreate(title=f"{needle} item {index}", description=None),
+            owner_id=user.id,
+        )
+        for index in range(3)
+    ]
+
+    response = client.get(
+        f"{settings.API_V1_STR}/search/",
+        headers=superuser_token_headers,
+        params={"q": needle, "limit": 2},
+    )
+
+    assert response.status_code == 200
+    content = response.json()
+    assert content["count"] >= 3
+    assert len(content["data"]) == 2
+    item_ids = []
+    for item in items:
+        assert item.id is not None
+        item_ids.append(item.id)
+    expected_ids = sorted(item_ids, reverse=True)[:2]
+    assert [result["id"] for result in content["data"]] == expected_ids
+
+
+def test_search_returns_empty_results_for_blank_query(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    response = client.get(
+        f"{settings.API_V1_STR}/search/",
+        headers=superuser_token_headers,
+        params={"q": "   "},
+    )
+
+    assert response.status_code == 200
+    assert response.json() == {"data": [], "count": 0}

frontend/src/client/models.ts

@@ -45,6 +45,22 @@ export type ItemsPublic = {
 
 
 
+export type SearchResult = {
+	type: 'item' | 'user';
+	id: number;
+	title: string;
+	description?: string | null;
+};
+
+
+
+export type SearchResultsPublic = {
+	data: Array<SearchResult>;
+	count: number;
+};
+
+
+
 export type Message = {
 	message: string;
 };
@@ -129,4 +145,3 @@ export type ValidationError = {
 	msg: string;
 	type: string;
 };
-

frontend/src/client/services.ts

@@ -2,7 +2,7 @@ import type { CancelablePromise } from './core/CancelablePromise';
 import { OpenAPI } from './core/OpenAPI';
 import { request as __request } from './core/request';
 
-import type { Body_login_login_access_token,Message,NewPassword,Token,UserPublic,UpdatePassword,UserCreate,UserRegister,UsersPublic,UserUpdate,UserUpdateMe,ItemCreate,ItemPublic,ItemsPublic,ItemUpdate } from './models';
+import type { Body_login_login_access_token,Message,NewPassword,Token,UserPublic,UpdatePassword,UserCreate,UserRegister,UsersPublic,UserUpdate,UserUpdateMe,ItemCreate,ItemPublic,ItemsPublic,ItemUpdate,SearchResultsPublic } from './models';
 
 export type TDataLoginAccessToken = {
                 formData: Body_login_login_access_token
@@ -521,4 +521,37 @@ id,
 		});
 	}
 
-}
+}
+
+export type TDataSearch = {
+                limit?: number
+q: string
+
+            }
+
+export class SearchService {
+
+	/**
+	 * Search
+	 * Search across resources visible to the current user.
+	 * @returns SearchResultsPublic Successful Response
+	 * @throws ApiError
+	 */
+	public static search(data: TDataSearch): CancelablePromise<SearchResultsPublic> {
+		const {
+limit = 10,
+q,
+} = data;
+		return __request(OpenAPI, {
+			method: 'GET',
+			url: '/api/v1/search/',
+			query: {
+				q, limit
+			},
+			errors: {
+				422: `Validation Error`,
+			},
+		});
+	}
+
+}