test: sim2 bookstore REST API extracted output (4 files)

2026-03-13 11:39:03 +05:30
parent 82078bd9e8
commit 9e89c031fd
4 changed files with 260 additions and 0 deletions
@@ -0,0 +1,15 @@
+# Book Management API
+
+A RESTful API for managing books built with FastAPI.
+
+## Features
+
+- Full CRUD operations for books
+- Input validation with Pydantic
+- Thread-safe in-memory storage
+- Comprehensive error handling
+- Interactive API documentation
+
+## Installation
+
+1. Install dependencies:
@@ -0,0 +1,175 @@
+from fastapi import FastAPI, HTTPException, status
+from fastapi.responses import JSONResponse
+from typing import List, Dict
+import threading
+from models import Book, BookCreate, BookUpdate
+
+# Thread-safe in-memory storage
+class BookStorage:
+    def __init__(self):
+        self._books: Dict[int, Book] = {}
+        self._next_id = 1
+        self._lock = threading.RLock()
+    
+    def create_book(self, book_data: BookCreate) -> Book:
+        with self._lock:
+            # Check for duplicate ISBN
+            for existing_book in self._books.values():
+                if existing_book.isbn == book_data.isbn:
+                    raise ValueError(f"Book with ISBN {book_data.isbn} already exists")
+            
+            book = Book(id=self._next_id, **book_data.dict())
+            self._books[self._next_id] = book
+            self._next_id += 1
+            return book
+    
+    def get_book(self, book_id: int) -> Book:
+        with self._lock:
+            if book_id not in self._books:
+                raise KeyError(f"Book with id {book_id} not found")
+            return self._books[book_id]
+    
+    def get_all_books(self) -> List[Book]:
+        with self._lock:
+            return list(self._books.values())
+    
+    def update_book(self, book_id: int, book_data: BookUpdate) -> Book:
+        with self._lock:
+            if book_id not in self._books:
+                raise KeyError(f"Book with id {book_id} not found")
+            
+            # Check for duplicate ISBN if updating ISBN
+            if book_data.isbn:
+                for existing_id, existing_book in self._books.items():
+                    if existing_id != book_id and existing_book.isbn == book_data.isbn:
+                        raise ValueError(f"Book with ISBN {book_data.isbn} already exists")
+            
+            current_book = self._books[book_id]
+            update_data = book_data.dict(exclude_unset=True)
+            updated_book = current_book.copy(update=update_data)
+            self._books[book_id] = updated_book
+            return updated_book
+    
+    def delete_book(self, book_id: int) -> bool:
+        with self._lock:
+            if book_id not in self._books:
+                raise KeyError(f"Book with id {book_id} not found")
+            del self._books[book_id]
+            return True
+
+# Initialize FastAPI app and storage
+app = FastAPI(
+    title="Book Management API",
+    description="A RESTful API for managing books with CRUD operations",
+    version="1.0.0"
+)
+
+book_storage = BookStorage()
+
+# Exception handlers
+@app.exception_handler(ValueError)
+async def value_error_handler(request, exc):
+    return JSONResponse(
+        status_code=status.HTTP_400_BAD_REQUEST,
+        content={"detail": str(exc)}
+    )
+
+@app.exception_handler(KeyError)
+async def key_error_handler(request, exc):
+    return JSONResponse(
+        status_code=status.HTTP_404_NOT_FOUND,
+        content={"detail": str(exc)}
+    )
+
+# API Endpoints
+@app.get("/books", response_model=List[Book], tags=["Books"])
+async def get_all_books():
+    """Retrieve all books"""
+    return book_storage.get_all_books()
+
+@app.get("/books/{book_id}", response_model=Book, tags=["Books"])
+async def get_book(book_id: int):
+    """Retrieve a specific book by ID"""
+    if book_id <= 0:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Book ID must be a positive integer"
+        )
+    
+    try:
+        return book_storage.get_book(book_id)
+    except KeyError as e:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(e)
+        )
+
+@app.post("/books", response_model=Book, status_code=status.HTTP_201_CREATED, tags=["Books"])
+async def create_book(book: BookCreate):
+    """Create a new book"""
+    try:
+        return book_storage.create_book(book)
+    except ValueError as e:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=str(e)
+        )
+
+@app.put("/books/{book_id}", response_model=Book, tags=["Books"])
+async def update_book(book_id: int, book: BookUpdate):
+    """Update an existing book"""
+    if book_id <= 0:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Book ID must be a positive integer"
+        )
+    
+    # Check if at least one field is provided for update
+    if not any(book.dict(exclude_unset=True).values()):
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="At least one field must be provided for update"
+        )
+    
+    try:
+        return book_storage.update_book(book_id, book)
+    except KeyError as e:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(e)
+        )
+    except ValueError as e:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=str(e)
+        )
+
+@app.delete("/books/{book_id}", status_code=status.HTTP_204_NO_CONTENT, tags=["Books"])
+async def delete_book(book_id: int):
+    """Delete a book by ID"""
+    if book_id <= 0:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Book ID must be a positive integer"
+        )
+    
+    try:
+        book_storage.delete_book(book_id)
+    except KeyError as e:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(e)
+        )
+
+@app.get("/", tags=["Root"])
+async def root():
+    """API root endpoint"""
+    return {
+        "message": "Book Management API",
+        "version": "1.0.0",
+        "docs": "/docs"
+    }
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)
@@ -0,0 +1,67 @@
+from pydantic import BaseModel, Field, validator
+from typing import Optional
+import re
+
+class BookBase(BaseModel):
+    title: str = Field(..., min_length=1, max_length=200, description="Book title")
+    author: str = Field(..., min_length=1, max_length=100, description="Book author")
+    price: float = Field(..., gt=0, description="Book price (must be positive)")
+    isbn: str = Field(..., min_length=10, max_length=17, description="ISBN-10 or ISBN-13")
+
+    @validator('isbn')
+    def validate_isbn(cls, v):
+        # Remove hyphens and spaces for validation
+        isbn_clean = re.sub(r'[-\s]', '', v)
+        
+        # Check if it's numeric and correct length
+        if not isbn_clean.isdigit():
+            raise ValueError('ISBN must contain only digits, hyphens, and spaces')
+        
+        if len(isbn_clean) not in [10, 13]:
+            raise ValueError('ISBN must be 10 or 13 digits long')
+        
+        return v
+
+    @validator('price')
+    def validate_price(cls, v):
+        # Round to 2 decimal places for currency
+        return round(v, 2)
+
+class BookCreate(BookBase):
+    pass
+
+class BookUpdate(BaseModel):
+    title: Optional[str] = Field(None, min_length=1, max_length=200)
+    author: Optional[str] = Field(None, min_length=1, max_length=100)
+    price: Optional[float] = Field(None, gt=0)
+    isbn: Optional[str] = Field(None, min_length=10, max_length=17)
+
+    @validator('isbn')
+    def validate_isbn(cls, v):
+        if v is not None:
+            isbn_clean = re.sub(r'[-\s]', '', v)
+            if not isbn_clean.isdigit():
+                raise ValueError('ISBN must contain only digits, hyphens, and spaces')
+            if len(isbn_clean) not in [10, 13]:
+                raise ValueError('ISBN must be 10 or 13 digits long')
+        return v
+
+    @validator('price')
+    def validate_price(cls, v):
+        if v is not None:
+            return round(v, 2)
+        return v
+
+class Book(BookBase):
+    id: int = Field(..., description="Unique book identifier")
+
+    class Config:
+        schema_extra = {
+            "example": {
+                "id": 1,
+                "title": "The Python Guide",
+                "author": "John Doe",
+                "price": 29.99,
+                "isbn": "978-0123456789"
+            }
+        }
@@ -0,0 +1,3 @@
+fastapi>=0.104.0
+uvicorn[standard]>=0.24.0
+pydantic>=2.0.0