a running example of a fastapi server

example-fastapi.py

# fully functional example of fastapi 
# # Create an item
# curl -X POST "http://localhost:8000/items/" -H "Content-Type: application/json" -d '{"id": 1, "name": "Test Item", "price": 9.99}'

# # Get all items
# curl "http://localhost:8000/items/"

# # Get specific item
# curl "http://localhost:8000/items/1"

# # Update an item
# curl -X PUT "http://localhost:8000/items/1" -H "Content-Type: application/json" -d '{"id": 1, "name": "Updated Item", "price": 19.99}'

# # Delete an item
# curl -X DELETE "http://localhost:8000/items/1"

# # Search items
# curl "http://localhost:8000/items/search/?name=test"

# # Health check
# curl "http://localhost:8000/health"


from fastapi import FastAPI, HTTPException, Query, Path
from pydantic import BaseModel, ConfigDict
from typing import Optional, List
import uvicorn

# Initialize FastAPI app
app = FastAPI(title="Sample FastAPI Server",
              description="A simple example of FastAPI functionality",
              version="1.0.0")

# Pydantic model for request/response validation
class Item(BaseModel):
    id: int
    name: str
    description: Optional[str] = None
    price: float
    is_available: bool = True

    class Config:
        arbitrary_types_allowed = True
        values = ConfigDict(allow_mutation=True)

# In-memory database
items_db = {}

# Root endpoint
@app.get("/")
async def root():
    return {"message": "Welcome to the FastAPI server!"}

# Create an item
@app.post("/items/", response_model=Item)
async def create_item(item: Item):
    if item.id in items_db:
        raise HTTPException(status_code=400, detail="Item ID already exists")
    items_db[item.id] = item
    return item

# Get all items
@app.get("/items/", response_model=List[Item])
async def get_items(skip: int = Query(0, ge=0), limit: int = Query(10, ge=1)):
    return list(items_db.values())

# Get a specific item by ID
@app.get("/items/{item_id}", response_model=Item)
async def get_item(item_id: int = Path(..., ge=0)):
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    # find the item by ID
    return items_db[item_id]

# Update an item
@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: int, item: Item):
    if item_id not in items_db.keys():
        raise HTTPException(status_code=404, detail="Item not found")
    if item_id != item.id:
        raise HTTPException(status_code=400, detail="Item ID mismatch")
    items_db[item_id] = item 
    return item

# Delete an item
@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    if item_id not in items_db.keys():
        raise HTTPException(status_code=404, detail="Item not found")
    del items_db[item_id]
    return {"message": "Item deleted successfully"}

# Search items by name
@app.get("/items/search/", response_model=List[Item])
async def search_items(name: str = Query(..., min_length=1)):
    matching_items = [] # TODO
    return matching_items

# Custom error handler
@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
    return {
        "status_code": exc.status_code,
        "detail": exc.detail
    }

# Health check endpoint
@app.get("/health")
async def health_check():
    return {"status": "healthy"}

# Run the server
if __name__ == "__main__":
    # Run the server using Uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)
    # uvicorn.run(app, host=