Coverage for / home / runner / work / bijux-cli / bijux-cli / src / bijux_cli / api / http.py: 99%
186 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-01-26 17:59 +0000
1# SPDX-License-Identifier: Apache-2.0
2# Copyright © 2025 Bijan Mousavi
4"""Provides a self-contained FastAPI application for a CRUD API.
6This module defines a complete HTTP API for managing "Item" resources using
7the FastAPI framework. It includes all necessary components for a functional
8web service.
10Services:
11 * **Pydantic Models:** `ItemIn`, `Item`, and response models for data
12 validation and serialization.
13 * **Storage Layer:** A formal `ItemStoreProtocol` and a concrete,
14 thread-safe `InMemoryItemStore` implementation.
15 * **API Endpoints:** A FastAPI `APIRouter` with path operations for all
16 CRUD (Create, Read, Update, Delete) actions.
17 * **Application Lifecycle:** A `lifespan` manager to prepopulate and
18 clear the data store on startup and shutdown.
19"""
21from __future__ import annotations
23from collections.abc import AsyncIterator, Awaitable, Callable
24from contextlib import asynccontextmanager
25import json
26import logging
27import threading
28from typing import Annotated, Any, Protocol, runtime_checkable
30from fastapi import (
31 APIRouter,
32 Body,
33 Depends,
34 FastAPI,
35 HTTPException,
36 Path,
37 Query,
38 Request,
39 Response,
40 status,
41)
42from fastapi.encoders import jsonable_encoder
43from fastapi.exceptions import RequestValidationError
44from fastapi.params import Depends as DependsMarker
45from fastapi.responses import JSONResponse
46from pydantic import AnyUrl, BaseModel, ConfigDict, Field, field_validator
48logger = logging.getLogger("bijux_cli.api.http")
51class Problem(BaseModel):
52 """Defines a standard RFC 7807 problem details response.
54 Attributes:
55 type (AnyUrl): A URI reference that identifies the problem type.
56 title (str): A short, human-readable summary of the problem type.
57 status (int): The HTTP status code.
58 detail (str): A human-readable explanation specific to this occurrence.
59 instance (str): A URI reference that identifies the specific occurrence.
60 """
62 type: AnyUrl = Field(
63 default=AnyUrl("about:blank"),
64 description="A URI reference that identifies the problem type.",
65 )
66 title: str = Field(..., description="A short, human-readable summary.")
67 status: int = Field(..., description="The HTTP status code.")
68 detail: str = Field(..., description="A human-readable explanation.")
69 instance: str = Field(..., description="A URI reference for this occurrence.")
72class ItemIn(BaseModel):
73 """Defines the input model for creating or updating an item.
75 Attributes:
76 name (str): The name of the item.
77 description (str | None): An optional description for the item.
78 """
80 model_config = ConfigDict(
81 extra="forbid",
82 )
84 name: str = Field(
85 ...,
86 min_length=1,
87 max_length=100,
88 json_schema_extra={"example": "Sample"},
89 )
90 description: str | None = Field(
91 None,
92 max_length=500,
93 json_schema_extra={"example": "Details about this item"},
94 )
96 @field_validator("name")
97 @classmethod
98 def validate_and_normalize_name(cls: type[ItemIn], v: str) -> str: # noqa: N805
99 """Strips whitespace and ensures the name is not empty.
101 Args:
102 v: The input string for the item's name.
104 Returns:
105 The validated and stripped name.
107 Raises:
108 ValueError: If the name is empty or contains only whitespace.
109 """
110 for ch in v:
111 codepoint = ord(ch)
112 if codepoint < 0x20 or 0x7F <= codepoint <= 0x9F:
113 raise ValueError("name must not contain control characters")
114 if 0xD800 <= codepoint <= 0xDFFF:
115 raise ValueError("name must not contain surrogate code points")
116 stripped_v = v.strip()
117 if not stripped_v:
118 raise ValueError("name must not be empty or contain only whitespace")
119 return stripped_v
122class Item(ItemIn):
123 """Defines the full item model, including its unique identifier.
125 Attributes:
126 id (int): The unique identifier for the item.
127 name (str): The name of the item.
128 description (str | None): An optional description for the item.
129 """
131 id: int = Field(..., json_schema_extra={"example": 1})
134class ItemListResponse(BaseModel):
135 """Defines the response model for a paginated list of items.
137 Attributes:
138 items (list[Item]): The list of items on the current page.
139 total (int): The total number of items available.
140 """
142 items: list[Item]
143 total: int
146@runtime_checkable
147class ItemStoreProtocol(Protocol):
148 """Defines the contract for an item storage service."""
150 def list_items(self, limit: int, offset: int) -> tuple[list[Item], int]:
151 """Lists items with pagination."""
152 ...
154 def get(self, item_id: int) -> Item:
155 """Gets an item by its unique ID."""
156 ...
158 def create(self, data: ItemIn) -> Item:
159 """Creates a new item."""
160 ...
162 def update(self, item_id: int, data: ItemIn) -> Item:
163 """Updates an existing item."""
164 ...
166 def delete(self, item_id: int) -> None:
167 """Deletes an item by its unique ID."""
168 ...
170 def reset(self) -> None:
171 """Resets the store to its initial empty state."""
172 ...
174 def prepopulate(self, data: list[dict[str, Any]]) -> None:
175 """Prepopulates the store with a list of items."""
176 ...
178 def find_by_name(self, name: str) -> Item | None:
179 """Returns an item by its name if it exists, otherwise None."""
180 ...
183class InMemoryItemStore(ItemStoreProtocol):
184 """A thread-safe, in-memory implementation of the `ItemStoreProtocol`.
186 Attributes:
187 _items (dict): The main dictionary storing items by their ID.
188 _name_index (dict): An index to enforce unique item names.
189 _lock (threading.RLock): A lock to ensure thread-safe operations.
190 _next_id (int): A counter for generating new item IDs.
191 """
193 def __init__(self) -> None:
194 """Initializes the in-memory item store."""
195 self._items: dict[int, Item] = {}
196 self._name_index: dict[str, int] = {}
197 self._lock = threading.RLock()
198 self._next_id = 1
200 def list_items(self, limit: int, offset: int) -> tuple[list[Item], int]:
201 """Lists items with pagination in a thread-safe manner.
203 Args:
204 limit (int): The maximum number of items to return.
205 offset (int): The starting index for the items to return.
207 Returns:
208 A tuple containing the list of items and the total number of items.
209 """
210 with self._lock:
211 items = list(self._items.values())
212 return items[offset : offset + limit], len(items)
214 def get(self, item_id: int) -> Item:
215 """Gets an item by its unique ID.
217 Args:
218 item_id (int): The ID of the item to retrieve.
220 Returns:
221 The requested item.
223 Raises:
224 HTTPException: With status 404 if the item is not found.
225 """
226 with self._lock:
227 item = self._items.get(item_id)
228 if not item:
229 raise HTTPException(
230 status_code=status.HTTP_404_NOT_FOUND,
231 detail=Problem(
232 type=AnyUrl("https://bijux-cli.dev/docs/errors/not-found"),
233 title="Not found",
234 status=status.HTTP_404_NOT_FOUND,
235 detail="Item not found",
236 instance=f"/v1/items/{item_id}",
237 ).model_dump(mode="json"),
238 )
239 return item
241 def create(self, data: ItemIn) -> Item:
242 """Creates a new item.
244 Args:
245 data (ItemIn): The data for the new item.
247 Returns:
248 The newly created item, including its generated ID.
250 Raises:
251 HTTPException: With status 409 if an item with the same name exists.
252 """
253 with self._lock:
254 key = data.name.strip().lower()
255 if key in self._name_index:
256 raise HTTPException(
257 status_code=status.HTTP_409_CONFLICT,
258 detail=Problem(
259 type=AnyUrl("https://bijux-cli.dev/docs/errors/conflict"),
260 title="Conflict",
261 status=status.HTTP_409_CONFLICT,
262 detail="Item with this name already exists",
263 instance="/v1/items",
264 ).model_dump(mode="json"),
265 )
266 item_id = self._next_id
267 self._next_id += 1
268 item = Item(id=item_id, name=data.name, description=data.description)
269 self._items[item_id] = item
270 self._name_index[key] = item_id
271 logger.info("Created item: %s", item)
272 return item
274 def update(self, item_id: int, data: ItemIn) -> Item:
275 """Update an existing item.
277 Args:
278 item_id (int): The unique identifier of the item to update.
279 data (ItemIn): The new values for the item.
281 Returns:
282 The updated item.
284 Raises:
285 HTTPException: If the item does not exist (HTTP 404) or if the new name
286 conflicts with another item (HTTP 409).
287 """
288 with self._lock:
289 existing = self._items.get(item_id)
290 if existing is None:
291 raise HTTPException(
292 status_code=status.HTTP_404_NOT_FOUND,
293 detail=Problem(
294 type=AnyUrl("https://bijux-cli.dev/docs/errors/not-found"),
295 title="Not found",
296 status=status.HTTP_404_NOT_FOUND,
297 detail="Item not found",
298 instance=f"/v1/items/{item_id}",
299 ).model_dump(mode="json"),
300 )
302 old_key = existing.name.strip().lower()
303 new_key = data.name.strip().lower()
305 if new_key != old_key and new_key in self._name_index:
306 raise HTTPException(
307 status_code=status.HTTP_409_CONFLICT,
308 detail=Problem(
309 type=AnyUrl("https://bijux-cli.dev/docs/errors/conflict"),
310 title="Conflict",
311 status=status.HTTP_409_CONFLICT,
312 detail="Item with this name already exists",
313 instance=f"/v1/items/{item_id}",
314 ).model_dump(mode="json"),
315 )
317 updated = Item(id=item_id, name=data.name, description=data.description)
318 self._items[item_id] = updated
319 if new_key != old_key: 319 ↛ 322line 319 didn't jump to line 322 because the condition on line 319 was always true
320 self._name_index.pop(old_key, None)
321 self._name_index[new_key] = item_id
322 logger.info("Updated item id=%s", item_id)
323 return updated
325 def delete(self, item_id: int) -> None:
326 """Delete an item by its unique ID.
328 Args:
329 item_id: The unique ID of the item to delete.
330 """
331 with self._lock:
332 existing = self._items.pop(item_id, None)
333 if existing is None:
334 raise HTTPException(
335 status_code=status.HTTP_404_NOT_FOUND,
336 detail=Problem(
337 type=AnyUrl("https://bijux-cli.dev/docs/errors/not-found"),
338 title="Not found",
339 status=status.HTTP_404_NOT_FOUND,
340 detail="Item not found",
341 instance=f"/v1/items/{item_id}",
342 ).model_dump(mode="json"),
343 )
344 self._name_index.pop(existing.name.strip().lower(), None)
345 logger.info("Deleted item id=%s", item_id)
347 def reset(self) -> None:
348 """Resets the store to its initial empty state."""
349 with self._lock:
350 self._items.clear()
351 self._name_index.clear()
352 self._next_id = 1
353 logger.info("Store reset")
355 def prepopulate(self, data: list[dict[str, Any]]) -> None:
356 """Prepopulates the store with a list of items.
358 Args:
359 data: A list of dictionaries, where each dictionary contains
360 the data for a new item.
361 """
362 with self._lock:
363 for entry in data:
364 self.create(ItemIn(**entry))
366 def find_by_name(self, name: str) -> Item | None:
367 """Lookup an item by its name (case-insensitive, trimmed)."""
368 with self._lock:
369 key = name.strip().lower()
370 item_id = self._name_index.get(key)
371 return self._items.get(item_id) if item_id is not None else None
374def get_store() -> ItemStoreProtocol:
375 """A FastAPI dependency to provide the `ItemStoreProtocol` instance."""
376 return store
379def get_item_or_404(
380 item_id: int = Path(..., ge=1),
381 store: ItemStoreProtocol = Depends(get_store), # noqa: B008
382) -> Item:
383 """A dependency that retrieves an item by ID or raises a 404."""
384 return store.get(item_id)
387def reject_duplicate_query_params(*params: str) -> DependsMarker:
388 """Create a dependency that rejects duplicate query parameters (HTTP 422).
390 Args:
391 *params: Names of query parameters that must not appear more than once.
393 Returns:
394 fastapi.params.Depends: A dependency marker that, when executed at
395 request time, raises ``HTTPException`` (422) if any listed parameter
396 appears more than once.
398 Raises:
399 HTTPException: Emitted at request time if duplicates are detected.
400 """
402 async def _dep(request: Request) -> None:
403 """Raise an HTTPException if specific query parameters are duplicated.
405 This function is designed to be used as a FastAPI dependency. It checks an
406 iterable of parameter names (assumed to be in the parent scope's `params`
407 variable) to ensure they are not repeated in the request's query string.
409 Args:
410 request: The incoming FastAPI/Starlette request object.
412 Raises:
413 HTTPException: An exception with a 422 status code and a
414 problem+json body if any of the specified query parameters
415 are found more than once.
416 """
417 duplicates = [p for p in params if len(request.query_params.getlist(p)) > 1]
418 if duplicates:
419 detail = f"Duplicate query params found: {', '.join(sorted(duplicates))}"
420 raise HTTPException(
421 status_code=422,
422 detail=Problem(
423 type=AnyUrl("https://bijux-cli.dev/docs/errors/validation-error"),
424 title="Validation error",
425 status=422,
426 detail=detail,
427 instance=str(request.url),
428 ).model_dump(mode="json"),
429 )
431 return DependsMarker(_dep)
434router = APIRouter(prefix="/v1")
437def require_accept_json(request: Request) -> None:
438 """Reject requests that don't accept application/json (HTTP 406).
440 Schemathesis' negative-data checks may send unsupported Accept headers.
441 If the client doesn't accept JSON (and not */*), respond with 406.
442 """
443 accept = request.headers.get("accept", "*/*").lower()
444 if "*/*" in accept or "application/json" in accept:
445 return
446 raise HTTPException(
447 status_code=status.HTTP_406_NOT_ACCEPTABLE,
448 detail=Problem(
449 type=AnyUrl("https://bijux-cli.dev/docs/errors/not-acceptable"),
450 title="Not Acceptable",
451 status=status.HTTP_406_NOT_ACCEPTABLE,
452 detail="Set 'Accept: application/json' for this endpoint",
453 instance=str(request.url),
454 ).model_dump(mode="json"),
455 )
458def allow_only(*allowed: str) -> Callable[[Request], Awaitable[None]]:
459 """Create a dependency that rejects unknown query parameters (HTTP 422).
461 Args:
462 *allowed (str): The set of query parameter names that are permitted.
464 Returns:
465 Callable[[Request], Awaitable[None]]: An async dependency suitable for
466 FastAPI's ``dependencies=[...]``. It raises an ``HTTPException`` with
467 status 422 if the request includes parameters outside the allowlist.
469 Raises:
470 HTTPException: Emitted by the returned dependency at request time when
471 unknown query parameters are present (422 Unprocessable Entity).
472 """
473 allowed_set: set[str] = set(allowed)
475 async def _dep(request: Request) -> None:
476 """Raise an HTTPException if unknown query parameters are present.
478 This function is designed to be used as a FastAPI dependency. It validates
479 that the request's query string contains only parameters from a pre-defined
480 set of allowed names, assumed to be in the parent scope's `allowed_set`
481 variable.
483 Args:
484 request: The incoming FastAPI/Starlette request object.
486 Raises:
487 HTTPException: An exception with a 422 status code and a
488 problem+json body if any query parameters are found that are
489 not in the `allowed_set`.
490 """
491 extras = set(request.query_params.keys()) - allowed_set
492 if extras:
493 raise HTTPException(
494 status_code=422,
495 detail=Problem(
496 type=AnyUrl("https://bijux-cli.dev/docs/errors/validation-error"),
497 title="Validation error",
498 status=422,
499 detail=f"Unknown query params: {', '.join(sorted(extras))}",
500 instance=str(request.url),
501 ).model_dump(mode="json"),
502 )
504 return _dep
507@router.get(
508 "/items",
509 response_model=ItemListResponse,
510 summary="List items",
511 description="List all items with pagination.",
512 tags=["Items"],
513 responses={406: {"model": Problem}, 422: {"model": Problem}},
514 dependencies=[
515 Depends(require_accept_json),
516 Depends(allow_only("limit", "offset")),
517 reject_duplicate_query_params("limit", "offset"),
518 ],
519)
520def list_items(
521 limit: int = Query(10, ge=1, le=100),
522 offset: int = Query(0, ge=0),
523 store: ItemStoreProtocol = Depends(get_store), # noqa: B008
524) -> ItemListResponse:
525 """Retrieves a paginated list of items.
527 Args:
528 limit (int): The maximum number of items per page.
529 offset (int): The starting offset for the item list.
530 store (ItemStoreProtocol): The dependency-injected item store.
532 Returns:
533 ItemListResponse: An object containing the list of items and total count.
534 """
535 items, total = store.list_items(limit, offset)
536 return ItemListResponse(items=items, total=total)
539@router.get(
540 "/items/{item_id}",
541 response_model=Item,
542 summary="Get item",
543 description="Get a single item by its ID.",
544 responses={404: {"model": Problem}, 406: {"model": Problem}},
545 tags=["Items"],
546 dependencies=[Depends(require_accept_json)],
547)
548def get_item(
549 item: Item = Depends(get_item_or_404), # noqa: B008
550) -> Item:
551 """Retrieves a single item by its ID.
553 This endpoint uses a dependency (`get_item_or_404`) to fetch the item,
554 ensuring that a 404 response is returned if the item does not exist.
556 Args:
557 item (Item): The item retrieved by the `get_item_or_404` dependency.
559 Returns:
560 Item: The requested item.
561 """
562 return item
565@router.post(
566 "/items",
567 response_model=Item,
568 status_code=status.HTTP_201_CREATED,
569 summary="Create item",
570 description="Create a new item.",
571 responses={
572 200: {
573 "model": Item,
574 "description": "Item already exists; existing resource returned",
575 },
576 406: {"model": Problem},
577 409: {"model": Problem},
578 422: {"model": Problem},
579 },
580 tags=["Items"],
581 dependencies=[Depends(require_accept_json)],
582)
583def create_item(
584 response: Response,
585 item: ItemIn = Body(...), # noqa: B008
586 store: ItemStoreProtocol = Depends(get_store), # noqa: B008
587) -> Item:
588 """Creates a new item.
590 Args:
591 item (ItemIn): The data for the new item from the request body.
592 store (ItemStoreProtocol): The dependency-injected item store.
594 Returns:
595 Item: The newly created item, including its server-generated ID.
596 """
597 existing = store.find_by_name(item.name)
598 if existing is not None:
599 response.status_code = status.HTTP_200_OK
600 return existing
601 return store.create(item)
604@router.put(
605 "/items/{item_id}",
606 response_model=Item,
607 summary="Update item",
608 description="Update an existing item.",
609 responses={
610 406: {"model": Problem},
611 404: {"model": Problem},
612 409: {"model": Problem},
613 422: {"model": Problem},
614 },
615 tags=["Items"],
616 dependencies=[Depends(require_accept_json)],
617)
618def update_item(
619 item: Annotated[Item, Depends(get_item_or_404)],
620 update_data: Annotated[ItemIn, Body(...)],
621 store: Annotated[ItemStoreProtocol, Depends(get_store)],
622) -> Item:
623 """Update an existing item.
625 Args:
626 item (Item): The current item resolved from the path parameter,
627 injected by ``get_item_or_404``.
628 update_data (ItemIn): The new values for the item (request body).
629 store (ItemStoreProtocol): The item store implementation (injected).
631 Returns:
632 The updated item.
634 Raises:
635 HTTPException: If the item does not exist (404) or if the new name
636 conflicts with another item (409). These are raised by the dependency
637 or the store layer.
638 RequestValidationError: If the path/body validation fails (422). Handled
639 by the global validation exception handler.
640 """
641 return store.update(item.id, update_data)
644@router.delete(
645 "/items/{item_id}",
646 status_code=status.HTTP_204_NO_CONTENT,
647 summary="Delete item",
648 description="Delete an item by its ID.",
649 responses={
650 406: {"model": Problem},
651 404: {"model": Problem},
652 422: {"model": Problem},
653 },
654 tags=["Items"],
655 dependencies=[Depends(require_accept_json)],
656)
657def delete_item(
658 item: Annotated[Item, Depends(get_item_or_404)],
659 store: Annotated[ItemStoreProtocol, Depends(get_store)],
660) -> Response:
661 """Delete an item by its unique ID.
663 The target item is resolved by the `get_item_or_404` dependency before this
664 handler runs.
666 Args:
667 item: The item to delete, injected by `get_item_or_404`.
668 store: The item store implementation, injected.
670 Returns:
671 Response: Empty body with **204 No Content** on successful deletion.
673 Raises:
674 HTTPException: 404 if the item does not exist (raised by the dependency).
675 """
676 store.delete(item.id)
677 return Response(status_code=status.HTTP_204_NO_CONTENT)
680@asynccontextmanager
681async def lifespan(app: FastAPI) -> AsyncIterator[None]:
682 """Manages the application's lifespan events for startup and shutdown.
684 On startup, this context manager resets and prepopulates the in-memory
685 store with demo data. On shutdown, it resets the store again.
687 Args:
688 app (FastAPI): The FastAPI application instance.
690 Yields:
691 None: Yields control to the application while it is running.
692 """
693 store.reset()
694 store.prepopulate(
695 [
696 {"name": "Item One", "description": "Description one"},
697 {"name": "Item Two", "description": "Description two"},
698 ]
699 )
700 logger.info("Prepopulated store with demo items")
701 yield
702 store.reset()
703 logger.info("Store reset on shutdown")
706store = InMemoryItemStore()
707app = FastAPI(
708 title="Bijux CLI API",
709 version="1.0.0",
710 description="High-quality demo API for educational/reference purposes.",
711 lifespan=lifespan,
712)
713app.include_router(router)
716@app.get("/health", summary="Health check", tags=["Health"])
717async def health() -> dict[str, str]:
718 """Lightweight readiness probe used by Makefile `api-test`."""
719 return {"status": "ok"}
722@app.exception_handler(RequestValidationError)
723async def validation_exception_handler(
724 request: Request, exc: RequestValidationError
725) -> JSONResponse:
726 """A custom exception handler for `RequestValidationError`.
728 This handler intercepts validation errors from FastAPI and formats them
729 into a standard `JSONResponse` with a 422 status code.
731 Args:
732 request (Request): The incoming request.
733 exc (RequestValidationError): The validation exception.
735 Returns:
736 JSONResponse: A JSON response detailing the validation error.
737 """
738 errors = jsonable_encoder(exc.errors())
739 logger.warning("Validation error: %s", errors)
740 return JSONResponse(
741 status_code=422,
742 content={
743 "type": "https://bijux-cli.dev/docs/errors/validation-error",
744 "title": "Validation error",
745 "status": 422,
746 "detail": json.dumps(errors),
747 "instance": str(request.url),
748 },
749 )
752@app.exception_handler(HTTPException)
753async def http_exception_handler(request: Request, exc: HTTPException) -> JSONResponse:
754 """A custom exception handler for `HTTPException`.
756 This handler intercepts FastAPI's standard HTTP exceptions and ensures they
757 are logged and returned in the standard JSON error format.
759 Args:
760 request (Request): The incoming request.
761 exc (HTTPException): The HTTP exception.
763 Returns:
764 JSONResponse: A JSON response detailing the HTTP error.
765 """
766 logger.warning("HTTP error: %s %s", exc.status_code, exc.detail)
767 return JSONResponse(status_code=exc.status_code, content=exc.detail)