forked from cleveragents/cleveragents-core
aditya
9fe3196827
All sims re-run after pipeline fixes (better prompts, real syntax validation, fenced-block extraction fix). Results: 7/7 PASS, 20 files total, all Python files pass compile() syntax check. Includes run_all_sims.py runner script and rag-basic action config.
120 lines
4.4 KiB
Python
120 lines
4.4 KiB
Python
from fastapi import FastAPI, HTTPException
|
|
from pydantic import BaseModel, Field
|
|
from typing import Dict, List, Optional
|
|
import uvicorn
|
|
|
|
app = FastAPI(
|
|
title="Bookstore API",
|
|
description="A simple REST API for managing books in a bookstore",
|
|
version="1.0.0"
|
|
)
|
|
|
|
# Pydantic models
|
|
class Book(BaseModel):
|
|
id: int = Field(..., description="Unique identifier for the book")
|
|
title: str = Field(..., min_length=1, description="Title of the book")
|
|
author: str = Field(..., min_length=1, description="Author of the book")
|
|
price: float = Field(..., gt=0, description="Price of the book in dollars")
|
|
isbn: str = Field(..., min_length=10, max_length=17, description="ISBN of the book")
|
|
|
|
class BookCreate(BaseModel):
|
|
title: str = Field(..., min_length=1, description="Title of the book")
|
|
author: str = Field(..., min_length=1, description="Author of the book")
|
|
price: float = Field(..., gt=0, description="Price of the book in dollars")
|
|
isbn: str = Field(..., min_length=10, max_length=17, description="ISBN of the book")
|
|
|
|
class BookUpdate(BaseModel):
|
|
title: Optional[str] = Field(None, min_length=1, description="Title of the book")
|
|
author: Optional[str] = Field(None, min_length=1, description="Author of the book")
|
|
price: Optional[float] = Field(None, gt=0, description="Price of the book in dollars")
|
|
isbn: Optional[str] = Field(None, min_length=10, max_length=17, description="ISBN of the book")
|
|
|
|
# In-memory storage
|
|
books_db: Dict[int, Book] = {}
|
|
next_book_id = 1
|
|
|
|
# Sample data
|
|
sample_books = [
|
|
{"title": "The Great Gatsby", "author": "F. Scott Fitzgerald", "price": 12.99, "isbn": "978-0-7432-7356-5"},
|
|
{"title": "To Kill a Mockingbird", "author": "Harper Lee", "price": 14.99, "isbn": "978-0-06-112008-4"},
|
|
{"title": "1984", "author": "George Orwell", "price": 13.99, "isbn": "978-0-452-28423-4"}
|
|
]
|
|
|
|
# Initialize sample data
|
|
for book_data in sample_books:
|
|
book = Book(id=next_book_id, **book_data)
|
|
books_db[next_book_id] = book
|
|
next_book_id += 1
|
|
|
|
@app.get("/")
|
|
async def root():
|
|
return {"message": "Welcome to the Bookstore API", "docs": "/docs"}
|
|
|
|
@app.get("/books", response_model=List[Book])
|
|
async def get_all_books():
|
|
"""Retrieve all books from the bookstore."""
|
|
return list(books_db.values())
|
|
|
|
@app.get("/books/{book_id}", response_model=Book)
|
|
async def get_book(book_id: int):
|
|
"""Retrieve a specific book by its ID."""
|
|
if book_id not in books_db:
|
|
raise HTTPException(status_code=404, detail="Book not found")
|
|
return books_db[book_id]
|
|
|
|
@app.post("/books", response_model=Book, status_code=201)
|
|
async def create_book(book: BookCreate):
|
|
"""Create a new book in the bookstore."""
|
|
global next_book_id
|
|
|
|
# Check if ISBN already exists
|
|
for existing_book in books_db.values():
|
|
if existing_book.isbn == book.isbn:
|
|
raise HTTPException(status_code=400, detail="Book with this ISBN already exists")
|
|
|
|
new_book = Book(id=next_book_id, **book.dict())
|
|
books_db[next_book_id] = new_book
|
|
next_book_id += 1
|
|
|
|
return new_book
|
|
|
|
@app.put("/books/{book_id}", response_model=Book)
|
|
async def update_book(book_id: int, book_update: BookUpdate):
|
|
"""Update an existing book by its ID."""
|
|
if book_id not in books_db:
|
|
raise HTTPException(status_code=404, detail="Book not found")
|
|
|
|
existing_book = books_db[book_id]
|
|
|
|
# Check if ISBN already exists for another book
|
|
if book_update.isbn:
|
|
for bid, existing in books_db.items():
|
|
if bid != book_id and existing.isbn == book_update.isbn:
|
|
raise HTTPException(status_code=400, detail="Book with this ISBN already exists")
|
|
|
|
# Update only provided fields
|
|
update_data = book_update.dict(exclude_unset=True)
|
|
updated_book_data = existing_book.dict()
|
|
updated_book_data.update(update_data)
|
|
|
|
updated_book = Book(**updated_book_data)
|
|
books_db[book_id] = updated_book
|
|
|
|
return updated_book
|
|
|
|
@app.delete("/books/{book_id}", status_code=204)
|
|
async def delete_book(book_id: int):
|
|
"""Delete a book from the bookstore by its ID."""
|
|
if book_id not in books_db:
|
|
raise HTTPException(status_code=404, detail="Book not found")
|
|
|
|
del books_db[book_id]
|
|
return None
|
|
|
|
# Health check endpoint
|
|
@app.get("/health")
|
|
async def health_check():
|
|
return {"status": "healthy", "total_books": len(books_db)}
|
|
|
|
if __name__ == "__main__":
|
|
uvicorn.run(app, host="0.0.0.0", port=8000) |