FastAPI Quick Start

Overview

Now that we have set up the development environment, it's time to create our first FastAPI application! This chapter will guide you through experiencing FastAPI's core functionality through actual code examples, including route creation, parameter handling, data validation, and automatic documentation generation.

🚀 First FastAPI Application

Simplest Hello World

Create main.py file:

from fastapi import FastAPI

# Create FastAPI application instance
app = FastAPI()

# Create route endpoint
@app.get("/")
async def read_root():
    return {"Hello": "World"}

# Run application
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Run the application:

# Method 1: Direct execution
python main.py

# Method 2: Use uvicorn (recommended)
uvicorn main:app --reload

# Method 3: Specify parameters
uvicorn main:app --host 0.0.0.0 --port 8000 --reload

Verify Application Running

# Test API
curl http://localhost:8000/

# Expected output: {"Hello":"World"}

Visit auto-generated documentation:

• Swagger UI: http://localhost:8000/docs
• ReDoc: http://localhost:8000/redoc

📊 Adding More Endpoints

Basic CRUD Operations

from fastapi import FastAPI
from typing import Optional

app = FastAPI(
    title="My First FastAPI App",
    description="First application to learn FastAPI",
    version="1.0.0"
)

# Mock database
fake_items_db = [
    {"item_id": 1, "name": "Laptop", "price": 1000, "description": "High-performance laptop"},
    {"item_id": 2, "name": "Mouse", "price": 25, "description": "Wireless mouse"},
    {"item_id": 3, "name": "Keyboard", "price": 75, "description": "Mechanical keyboard"}
]

# GET: Get all items
@app.get("/")
async def read_root():
    return {"message": "Welcome to FastAPI!", "items_count": len(fake_items_db)}

# GET: Get all items (with query parameters)
@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    return fake_items_db[skip: skip + limit]

# GET: Get single item
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: Optional[str] = None):
    for item in fake_items_db:
        if item["item_id"] == item_id:
            if q:
                return {"item": item, "query": q}
            return {"item": item}
    return {"error": "Item not found"}

# POST: Create new item
@app.post("/items/")
async def create_item(name: str, price: float, description: Optional[str] = None):
    new_item_id = max([item["item_id"] for item in fake_items_db]) + 1
    new_item = {
        "item_id": new_item_id,
        "name": name,
        "price": price,
        "description": description or ""
    }
    fake_items_db.append(new_item)
    return {"message": "Item created", "item": new_item}

# PUT: Update item
@app.put("/items/{item_id}")
async def update_item(item_id: int, name: str, price: float, description: Optional[str] = None):
    for i, item in enumerate(fake_items_db):
        if item["item_id"] == item_id:
            fake_items_db[i] = {
                "item_id": item_id,
                "name": name,
                "price": price,
                "description": description or item["description"]
            }
            return {"message": "Item updated", "item": fake_items_db[i]}
    return {"error": "Item not found"}

# DELETE: Delete item
@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    for i, item in enumerate(fake_items_db):
        if item["item_id"] == item_id:
            deleted_item = fake_items_db.pop(i)
            return {"message": "Item deleted", "item": deleted_item}
    return {"error": "Item not found"}

🎯 Using Pydantic Models

Define Data Models

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime

app = FastAPI(title="Product Management API", version="1.0.0")

# Pydantic model definitions
class ItemBase(BaseModel):
    name: str = Field(..., min_length=1, max_length=100, description="Item name")
    price: float = Field(..., gt=0, description="Item price, must be greater than 0")
    description: Optional[str] = Field(None, max_length=500, description="Item description")
    is_available: bool = Field(True, description="Availability status")

class ItemCreate(ItemBase):
    pass

class ItemUpdate(BaseModel):
    name: Optional[str] = Field(None, min_length=1, max_length=100)
    price: Optional[float] = Field(None, gt=0)
    description: Optional[str] = Field(None, max_length=500)
    is_available: Optional[bool] = None

class ItemResponse(ItemBase):
    item_id: int = Field(..., description="Item ID")
    created_at: datetime = Field(..., description="Creation time")
    updated_at: datetime = Field(..., description="Update time")

    class Config:
        # Allow creation from ORM objects
        from_attributes = True

# Mock database
fake_items_db: List[dict] = []
next_item_id = 1

# Helper functions
def get_current_time():
    return datetime.now()

def find_item_by_id(item_id: int):
    for item in fake_items_db:
        if item["item_id"] == item_id:
            return item
    return None

# API endpoints
@app.get("/", summary="Root path", description="Returns API basic information")
async def read_root():
    return {
        "message": "Product Management API",
        "version": "1.0.0",
        "items_count": len(fake_items_db)
    }

@app.get("/items/", response_model=List[ItemResponse], summary="Get all items")
async def read_items(
    skip: int = Field(0, ge=0, description="Number of records to skip"),
    limit: int = Field(10, ge=1, le=100, description="Number of records to return")
):
    """
    Get item list

    - **skip**: Number of records to skip (for pagination)
    - **limit**: Number of records to return (1-100)
    """
    return fake_items_db[skip: skip + limit]

@app.get("/items/{item_id}", response_model=ItemResponse, summary="Get single item")
async def read_item(item_id: int = Field(..., description="Item ID")):
    """
    Get item details by ID

    - **item_id**: Unique identifier of the item
    """
    item = find_item_by_id(item_id)
    if item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@app.post("/items/", response_model=ItemResponse, status_code=201, summary="Create new item")
async def create_item(item: ItemCreate):
    """
    Create new item

    - **name**: Item name (required, 1-100 characters)
    - **price**: Item price (required, greater than 0)
    - **description**: Item description (optional, max 500 characters)
    - **is_available**: Availability status (defaults to true)
    """
    global next_item_id

    current_time = get_current_time()
    new_item = {
        "item_id": next_item_id,
        "name": item.name,
        "price": item.price,
        "description": item.description,
        "is_available": item.is_available,
        "created_at": current_time,
        "updated_at": current_time
    }

    fake_items_db.append(new_item)
    next_item_id += 1

    return new_item

@app.put("/items/{item_id}", response_model=ItemResponse, summary="Update item")
async def update_item(item_id: int, item: ItemUpdate):
    """
    Update existing item

    - **item_id**: ID of the item to update
    - Only updates provided fields, other fields remain unchanged
    """
    existing_item = find_item_by_id(item_id)
    if existing_item is None:
        raise HTTPException(status_code=404, detail="Item not found")

    # Update fields
    update_data = item.dict(exclude_unset=True)
    for field, value in update_data.items():
        existing_item[field] = value

    existing_item["updated_at"] = get_current_time()

    return existing_item

@app.delete("/items/{item_id}", summary="Delete item")
async def delete_item(item_id: int):
    """
    Delete item

    - **item_id**: ID of the item to delete
    """
    for i, item in enumerate(fake_items_db):
        if item["item_id"] == item_id:
            deleted_item = fake_items_db.pop(i)
            return {"message": "Item deleted", "item": deleted_item}

    raise HTTPException(status_code=404, detail="Item not found")

🧪 API Testing

Testing with curl

# 1. Get all items
curl -X GET "http://localhost:8000/items/" -H "accept: application/json"

# 2. Create new item
curl -X POST "http://localhost:8000/items/" \
  -H "accept: application/json" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "MacBook Pro",
    "price": 2500.00,
    "description": "Apple MacBook Pro 16-inch",
    "is_available": true
  }'

# 3. Get single item
curl -X GET "http://localhost:8000/items/1" -H "accept: application/json"

# 4. Update item
curl -X PUT "http://localhost:8000/items/1" \
  -H "accept: application/json" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "MacBook Pro M2",
    "price": 2800.00
  }'

# 5. Delete item
curl -X DELETE "http://localhost:8000/items/1" -H "accept: application/json"

Testing with Python requests

Create test_api.py:

import requests
import json

# API base URL
BASE_URL = "http://localhost:8000"

def test_api():
    # 1. Test root path
    response = requests.get(f"{BASE_URL}/")
    print("Root path:", response.json())

    # 2. Create item
    new_item = {
        "name": "iPhone 15",
        "price": 999.00,
        "description": "Latest iPhone",
        "is_available": True
    }

    response = requests.post(f"{BASE_URL}/items/", json=new_item)
    print("Create item:", response.json())
    created_item = response.json()
    item_id = created_item["item_id"]

    # 3. Get item
    response = requests.get(f"{BASE_URL}/items/{item_id}")
    print("Get item:", response.json())

    # 4. Update item
    update_data = {
        "price": 1099.00,
        "description": "iPhone 15 Pro Max"
    }

    response = requests.put(f"{BASE_URL}/items/{item_id}", json=update_data)
    print("Update item:", response.json())

    # 5. Get all items
    response = requests.get(f"{BASE_URL}/items/")
    print("All items:", response.json())

    # 6. Delete item
    response = requests.delete(f"{BASE_URL}/items/{item_id}")
    print("Delete item:", response.json())

if __name__ == "__main__":
    test_api()

Run tests:

python test_api.py

📚 Auto Documentation Exploration

Swagger UI Features

Visit http://localhost:8000/docs, you can:

1. View all API endpoints
2. Test API online: Click "Try it out" button
3. View request/response models
4. Download OpenAPI specification

ReDoc Documentation

Visit http://localhost:8000/redoc for more beautiful documentation display.

OpenAPI JSON

Visit http://localhost:8000/openapi.json to get the raw OpenAPI specification.

🔧 Error Handling and Status Codes

Add Custom Exception Handlers

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()

# Custom exception class
class ItemNotFoundError(Exception):
    def __init__(self, item_id: int):
        self.item_id = item_id

# Exception handler
@app.exception_handler(ItemNotFoundError)
async def item_not_found_handler(request: Request, exc: ItemNotFoundError):
    return JSONResponse(
        status_code=404,
        content={
            "error": "Item not found",
            "message": f"Item with id {exc.item_id} does not exist",
            "type": "ItemNotFoundError"
        }
    )

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={
            "error": "Validation error",
            "message": "Input data validation failed",
            "details": exc.errors()
        }
    )

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": "HTTP error",
            "message": exc.detail,
            "status_code": exc.status_code
        }
    )

# Endpoint using custom exception
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    item = find_item_by_id(item_id)
    if item is None:
        raise ItemNotFoundError(item_id)
    return item

🎛️ Configuration and Environment

Environment Variable Configuration

Create .env file:

APP_NAME=FastAPI Learning Application
APP_VERSION=1.0.0
DEBUG=True
API_PREFIX=/api/v1

Configuration management:

from pydantic import BaseSettings

class Settings(BaseSettings):
    app_name: str = "FastAPI App"
    app_version: str = "1.0.0"
    debug: bool = False
    api_prefix: str = "/api/v1"

    class Config:
        env_file = ".env"

settings = Settings()

app = FastAPI(
    title=settings.app_name,
    version=settings.app_version,
    debug=settings.debug
)

# Use API prefix
from fastapi import APIRouter

api_router = APIRouter(prefix=settings.api_prefix)

@api_router.get("/items/")
async def read_items():
    return {"items": []}

app.include_router(api_router)

🚀 Production Ready Configuration

Complete Application Configuration

from fastapi import FastAPI, Depends
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
import time
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Create application
app = FastAPI(
    title="Production-grade FastAPI Application",
    description="FastAPI application with middleware and security configuration",
    version="1.0.0",
    docs_url="/api/docs",  # Custom documentation path
    redoc_url="/api/redoc",
    openapi_url="/api/openapi.json"
)

# Add CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "https://yourdomain.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Add trusted host middleware
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["localhost", "127.0.0.1", "yourdomain.com"]
)

# Custom middleware: request time logging
@app.middleware("http")
async def add_process_time_header(request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    logger.info(f"Request processed in {process_time:.4f} seconds")
    return response

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

# Startup and shutdown events
@app.on_event("startup")
async def startup_event():
    logger.info("Application startup complete")

@app.on_event("shutdown")
async def shutdown_event():
    logger.info("Application shutdown complete")

📋 Project Organization

Recommended File Structure

my_fastapi_app/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI application entry point
│   ├── config.py            # Configuration management
│   ├── models/              # Pydantic models
│   │   ├── __init__.py
│   │   └── item.py
│   ├── routers/             # Route modules
│   │   ├── __init__.py
│   │   └── items.py
│   ├── services/            # Business logic
│   │   ├── __init__.py
│   │   └── item_service.py
│   └── utils/               # Utility functions
│       ├── __init__.py
│       └── helpers.py
├── tests/                   # Test code
│   ├── __init__.py
│   └── test_main.py
├── requirements.txt         # Dependencies
├── .env                     # Environment variables
└── README.md               # Project description

Modular Code Example

app/models/item.py:

from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime

class ItemBase(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    price: float = Field(..., gt=0)
    description: Optional[str] = Field(None, max_length=500)

class ItemCreate(ItemBase):
    pass

class ItemUpdate(BaseModel):
    name: Optional[str] = Field(None, min_length=1, max_length=100)
    price: Optional[float] = Field(None, gt=0)
    description: Optional[str] = Field(None, max_length=500)

class ItemResponse(ItemBase):
    item_id: int
    created_at: datetime
    updated_at: datetime

    class Config:
        from_attributes = True

app/routers/items.py:

from fastapi import APIRouter, HTTPException
from typing import List
from app.models.item import ItemCreate, ItemUpdate, ItemResponse

router = APIRouter(prefix="/items", tags=["items"])

@router.get("/", response_model=List[ItemResponse])
async def read_items():
    return []

@router.post("/", response_model=ItemResponse, status_code=201)
async def create_item(item: ItemCreate):
    # Business logic
    pass

app/main.py:

from fastapi import FastAPI
from app.routers import items

app = FastAPI(title="Modular FastAPI Application")

# Include routes
app.include_router(items.router, prefix="/api/v1")

@app.get("/")
async def root():
    return {"message": "Modular FastAPI Application"}

Summary

In this chapter, we experienced FastAPI's core functionality through actual code:

• ✅ Create Basic API: Understand basic structure of FastAPI applications
• ✅ Routes and Endpoints: Implement CRUD operations
• ✅ Pydantic Models: Data validation and serialization
• ✅ Auto Documentation: Swagger UI and ReDoc
• ✅ Error Handling: Exception handling and status codes
• ✅ Middleware Configuration: CORS, logging, security
• ✅ Project Organization: Modularization and best practices

Now you have mastered the basic usage of FastAPI. In the next chapter, we will learn the routing system and HTTP method handling in depth.

Best Practices

• Always use Pydantic models for data validation
• Write clear docstrings for each endpoint
• Use appropriate HTTP status codes
• Organize code structure reasonably, keep it modular
• Utilize FastAPI's auto documentation for API testing

In the next chapter, we will learn FastAPI's routing system and advanced HTTP method handling.