🚀 FastAPI: Modern APIs

🎯 Introduction

Welcome to the world of FastAPI - the modern, fast, and incredibly fun way to build APIs in Python! 🎉

Imagine you’re building the backend for the next big app - maybe a food delivery service 🍕, a social media platform 📱, or a game leaderboard 🎮. You need something that’s:

• Lightning fast ⚡
• Easy to write and maintain 📝
• Automatically generates documentation 📚
• Validates data for you 🛡️

That’s exactly what FastAPI delivers! Built on modern Python features like type hints and async/await, it’s like having a supercharged API framework that does the heavy lifting for you. Let’s dive in and build something awesome! 🚀

📚 Understanding FastAPI

Think of FastAPI as your smart API assistant 🤖. While traditional frameworks are like manual cars (you control everything), FastAPI is like a Tesla with autopilot - it handles many things automatically while giving you full control when needed.

What Makes FastAPI Special? ✨

# Traditional Flask approach
@app.route('/user/<user_id>')
def get_user(user_id):
    # Hope user_id is valid... 🤞
    # Manually validate everything
    # Write your own docs
    return {"user_id": user_id}

# FastAPI approach 🚀
@app.get('/user/{user_id}')
def get_user(user_id: int):  # 👈 Automatic validation!
    # user_id is guaranteed to be an int
    # Docs generated automatically
    # Type hints = superpowers!
    return {"user_id": user_id}

FastAPI uses Python’s type hints to:

• Validate request data automatically 🛡️
• Serialize responses properly 📦
• Generate interactive API docs 📚
• Provide editor support and autocomplete 💡

🔧 Basic Syntax and Usage

Let’s build our first FastAPI application - a simple pizza ordering API! 🍕

Installing FastAPI

# Install FastAPI and an ASGI server
pip install fastapi uvicorn

Your First API

from fastapi import FastAPI
from pydantic import BaseModel
from typing import Optional

# 🎨 Create your FastAPI app
app = FastAPI(title="Pizza Palace API 🍕")

# 📝 Define data models with Pydantic
class Pizza(BaseModel):
    name: str
    size: str  # small, medium, large
    toppings: list[str]
    extra_cheese: Optional[bool] = False

# 🏠 Home route
@app.get("/")
def read_root():
    return {"message": "Welcome to Pizza Palace! 🍕"}

# 📋 Get menu
@app.get("/menu")
def get_menu():
    return {
        "pizzas": ["Margherita", "Pepperoni", "Hawaiian"],
        "sizes": ["small", "medium", "large"],
        "toppings": ["mushrooms", "olives", "peppers"]
    }

# 🍕 Order a pizza
@app.post("/order")
def create_order(pizza: Pizza):  # 👈 Automatic validation!
    # FastAPI validates the request body matches Pizza model
    price = calculate_price(pizza)
    return {
        "order": pizza,
        "price": price,
        "status": "preparing 👨‍🍳"
    }

def calculate_price(pizza: Pizza):
    base_prices = {"small": 10, "medium": 15, "large": 20}
    price = base_prices[pizza.size]
    price += len(pizza.toppings) * 1.5
    if pizza.extra_cheese:
        price += 2
    return price

Running Your API

# Start the server 🚀
uvicorn main:app --reload

# Your API is now live at http://localhost:8000
# Interactive docs at http://localhost:8000/docs 📚

💡 Practical Examples

Example 1: Building a Task Manager API 📝

Let’s create a more complex API for managing tasks:

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

app = FastAPI(title="TaskMaster API 📋")

# 🎯 Define enums for better validation
class Priority(str, Enum):
    low = "low"
    medium = "medium"
    high = "high"
    urgent = "urgent"

# 📦 Task model with validation
class Task(BaseModel):
    title: str = Field(..., min_length=1, max_length=100)
    description: Optional[str] = None
    priority: Priority = Priority.medium
    completed: bool = False
    due_date: Optional[datetime] = None
    
    class Config:
        schema_extra = {
            "example": {
                "title": "Learn FastAPI",
                "description": "Complete the FastAPI tutorial",
                "priority": "high",
                "due_date": "2024-12-31T23:59:59"
            }
        }

# 🗄️ In-memory database (for demo)
tasks_db = {}
task_counter = 0

# 📋 Get all tasks
@app.get("/tasks", response_model=List[Task])
def get_tasks(
    priority: Optional[Priority] = None,
    completed: Optional[bool] = None
):
    """Get all tasks with optional filters 🔍"""
    tasks = list(tasks_db.values())
    
    # Apply filters if provided
    if priority:
        tasks = [t for t in tasks if t.priority == priority]
    if completed is not None:
        tasks = [t for t in tasks if t.completed == completed]
    
    return tasks

# ➕ Create a task
@app.post("/tasks", response_model=dict, status_code=status.HTTP_201_CREATED)
def create_task(task: Task):
    """Create a new task 🎨"""
    global task_counter
    task_counter += 1
    task_id = f"task_{task_counter}"
    tasks_db[task_id] = task
    
    return {
        "id": task_id,
        "task": task,
        "message": "Task created successfully! 🎉"
    }

# 🔍 Get specific task
@app.get("/tasks/{task_id}", response_model=Task)
def get_task(task_id: str):
    """Get a specific task by ID 🎯"""
    if task_id not in tasks_db:
        raise HTTPException(
            status_code=404, 
            detail=f"Task {task_id} not found 😔"
        )
    return tasks_db[task_id]

# ✅ Complete a task
@app.patch("/tasks/{task_id}/complete")
def complete_task(task_id: str):
    """Mark a task as completed ✅"""
    if task_id not in tasks_db:
        raise HTTPException(
            status_code=404,
            detail=f"Task {task_id} not found 😔"
        )
    
    tasks_db[task_id].completed = True
    return {"message": "Task completed! Great job! 🎊"}

Example 2: Game Leaderboard API 🎮

from fastapi import FastAPI, Query
from pydantic import BaseModel, validator
from typing import List, Optional
import uuid

app = FastAPI(title="Game Leaderboard API 🏆")

# 🎮 Player score model
class PlayerScore(BaseModel):
    player_name: str
    score: int
    game_mode: str
    
    @validator('score')
    def score_must_be_positive(cls, v):
        if v < 0:
            raise ValueError('Score must be positive! 🚫')
        return v
    
    @validator('player_name')
    def name_must_be_valid(cls, v):
        if len(v) < 3:
            raise ValueError('Name must be at least 3 characters! 📝')
        if len(v) > 20:
            raise ValueError('Name too long! Max 20 characters 📏')
        return v

# 🏆 Leaderboard entry
class LeaderboardEntry(BaseModel):
    rank: int
    player_name: str
    score: int
    game_mode: str
    achievement: Optional[str] = None

# 🗄️ Store scores
leaderboard_db = []

# 🎯 Submit a score
@app.post("/score/submit")
def submit_score(score: PlayerScore):
    """Submit a new high score 🚀"""
    # Add to leaderboard
    leaderboard_db.append(score)
    
    # Sort by score (highest first)
    leaderboard_db.sort(key=lambda x: x.score, reverse=True)
    
    # Find player's rank
    rank = next(
        i for i, s in enumerate(leaderboard_db, 1) 
        if s.player_name == score.player_name and s.score == score.score
    )
    
    # Award achievements 🏅
    achievement = None
    if score.score >= 10000:
        achievement = "Legend 🌟"
    elif score.score >= 5000:
        achievement = "Master 🎯"
    elif score.score >= 1000:
        achievement = "Pro 💪"
    
    return {
        "rank": rank,
        "achievement": achievement,
        "message": f"Score submitted! You're rank #{rank}! 🎉"
    }

# 🏆 Get leaderboard
@app.get("/leaderboard", response_model=List[LeaderboardEntry])
def get_leaderboard(
    game_mode: Optional[str] = None,
    top: int = Query(default=10, ge=1, le=100)
):
    """Get the top players 🏆"""
    # Filter by game mode if specified
    scores = leaderboard_db
    if game_mode:
        scores = [s for s in scores if s.game_mode == game_mode]
    
    # Create leaderboard entries
    entries = []
    for i, score in enumerate(scores[:top], 1):
        # Award achievements
        achievement = None
        if score.score >= 10000:
            achievement = "Legend 🌟"
        elif score.score >= 5000:
            achievement = "Master 🎯"
        elif score.score >= 1000:
            achievement = "Pro 💪"
            
        entries.append(LeaderboardEntry(
            rank=i,
            player_name=score.player_name,
            score=score.score,
            game_mode=score.game_mode,
            achievement=achievement
        ))
    
    return entries

# 🎮 Get player stats
@app.get("/player/{player_name}/stats")
def get_player_stats(player_name: str):
    """Get statistics for a specific player 📊"""
    player_scores = [s for s in leaderboard_db if s.player_name == player_name]
    
    if not player_scores:
        return {"message": "No scores found for this player 😔"}
    
    return {
        "player": player_name,
        "total_games": len(player_scores),
        "highest_score": max(s.score for s in player_scores),
        "average_score": sum(s.score for s in player_scores) // len(player_scores),
        "favorite_mode": max(
            set(s.game_mode for s in player_scores),
            key=lambda m: sum(1 for s in player_scores if s.game_mode == m)
        )
    }

🚀 Advanced Concepts

Async Support for Lightning Speed ⚡

FastAPI fully supports async/await for maximum performance:

import asyncio
import httpx
from fastapi import FastAPI

app = FastAPI()

# 🏎️ Async endpoints for concurrent operations
@app.get("/weather/{city}")
async def get_weather(city: str):
    """Get weather data asynchronously 🌤️"""
    async with httpx.AsyncClient() as client:
        # Make multiple API calls concurrently
        weather_task = client.get(f"https://api.weather.com/{city}")
        forecast_task = client.get(f"https://api.weather.com/{city}/forecast")
        
        # Wait for both to complete
        weather, forecast = await asyncio.gather(
            weather_task,
            forecast_task
        )
    
    return {
        "city": city,
        "current": weather.json(),
        "forecast": forecast.json()
    }

# 🗄️ Async database operations
from typing import List
from sqlalchemy.ext.asyncio import AsyncSession

@app.get("/users/active")
async def get_active_users(db: AsyncSession):
    """Get all active users asynchronously 👥"""
    # Async database query
    result = await db.execute(
        "SELECT * FROM users WHERE active = true"
    )
    users = result.fetchall()
    
    # Process in parallel
    tasks = [process_user(user) for user in users]
    processed_users = await asyncio.gather(*tasks)
    
    return processed_users

Dependency Injection Magic 🪄

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

# 🔐 Authentication dependency
async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security)
):
    """Verify and return current user 👤"""
    token = credentials.credentials
    
    # Verify token (simplified)
    if token != "secret-token":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials 🚫"
        )
    
    return {"username": "player1", "user_id": 123}

# 🎮 Use dependencies in endpoints
@app.post("/game/save")
async def save_game(
    save_data: dict,
    current_user: dict = Depends(get_current_user)
):
    """Save game progress (authenticated) 💾"""
    return {
        "message": f"Game saved for {current_user['username']}! 🎮",
        "save_id": "save_12345"
    }

# 🏗️ Dependency chains
async def get_db():
    """Database connection dependency 🗄️"""
    db = AsyncDatabase()
    try:
        yield db
    finally:
        await db.close()

async def get_game_service(
    db = Depends(get_db),
    user = Depends(get_current_user)
):
    """Game service with dependencies 🎯"""
    return GameService(db, user)

@app.get("/game/stats")
async def get_game_stats(
    service: GameService = Depends(get_game_service)
):
    """Get game statistics using service 📊"""
    return await service.get_user_stats()

Background Tasks 🏃‍♂️

from fastapi import BackgroundTasks
import time

# 📧 Email notification function
def send_email_notification(email: str, message: str):
    """Simulate sending email (runs in background) 📬"""
    time.sleep(2)  # Simulate email sending
    print(f"Email sent to {email}: {message}")

@app.post("/order/pizza")
async def order_pizza(
    pizza: Pizza,
    email: str,
    background_tasks: BackgroundTasks
):
    """Order pizza with email confirmation 🍕"""
    # Process order immediately
    order_id = f"order_{int(time.time())}"
    
    # Schedule email notification for background
    background_tasks.add_task(
        send_email_notification,
        email,
        f"Your pizza order {order_id} is confirmed! 🍕"
    )
    
    # Return immediately (email sends in background)
    return {
        "order_id": order_id,
        "status": "confirmed",
        "message": "Check your email for confirmation! 📧"
    }

⚠️ Common Pitfalls and Solutions

❌ Wrong: Blocking operations in async endpoints

# ❌ BAD: Blocks the event loop
@app.get("/slow")
async def slow_endpoint():
    time.sleep(5)  # 😱 Blocks everything!
    return {"message": "Finally done..."}

✅ Right: Use async operations or run in executor

# ✅ GOOD: Non-blocking async sleep
@app.get("/fast")
async def fast_endpoint():
    await asyncio.sleep(5)  # 😊 Non-blocking!
    return {"message": "Done efficiently!"}

# ✅ GOOD: Run blocking code in executor
import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor()

@app.get("/compute")
async def compute_endpoint():
    # Run blocking operation in thread pool
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(
        executor,
        expensive_computation  # Blocking function
    )
    return {"result": result}

❌ Wrong: Not handling validation errors properly

# ❌ BAD: Crashes on invalid data
@app.post("/user")
def create_user(data: dict):
    age = int(data["age"])  # 💥 Crashes if age isn't a number
    return {"age": age}

✅ Right: Use Pydantic models for validation

# ✅ GOOD: Automatic validation with helpful errors
from pydantic import BaseModel, validator

class User(BaseModel):
    name: str
    age: int
    email: str
    
    @validator('age')
    def age_must_be_valid(cls, v):
        if v < 0 or v > 150:
            raise ValueError('Age must be between 0 and 150! 🎂')
        return v

@app.post("/user")
def create_user(user: User):  # 🛡️ Validated automatically!
    return {"message": f"User {user.name} created!"}

🛠️ Best Practices

1. Use Type Hints Everywhere 🎯

# ✅ Type hints = automatic validation + documentation
@app.get("/items/{item_id}", response_model=Item)
async def get_item(
    item_id: int,  # Path parameter
    q: Optional[str] = Query(None, max_length=50),  # Query parameter
    user: User = Depends(get_current_user)  # Dependency
) -> Item:  # Return type
    return await fetch_item(item_id, q, user)

2. Structure Your Project 📁

# project/
# ├── app/
# │   ├── __init__.py
# │   ├── main.py          # FastAPI app
# │   ├── models.py        # Pydantic models
# │   ├── routers/         # API routes
# │   │   ├── users.py
# │   │   └── items.py
# │   ├── services/        # Business logic
# │   └── dependencies.py  # Shared dependencies

# main.py
from fastapi import FastAPI
from app.routers import users, items

app = FastAPI(title="My Awesome API 🚀")

# Include routers
app.include_router(users.router, prefix="/users", tags=["users"])
app.include_router(items.router, prefix="/items", tags=["items"])

3. Handle Errors Gracefully 🛡️

from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse

# Custom exception handler
@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    return JSONResponse(
        status_code=400,
        content={
            "message": str(exc),
            "type": "validation_error",
            "tip": "Check your input values! 🔍"
        }
    )

# Global error handling
@app.middleware("http")
async def catch_exceptions(request: Request, call_next):
    try:
        return await call_next(request)
    except Exception as e:
        return JSONResponse(
            status_code=500,
            content={
                "message": "Something went wrong! 😰",
                "type": "server_error"
            }
        )

4. Use Environment Variables 🔐

from pydantic import BaseSettings

class Settings(BaseSettings):
    app_name: str = "FastAPI App"
    admin_email: str
    database_url: str
    secret_key: str
    
    class Config:
        env_file = ".env"

# Create settings instance
settings = Settings()

# Use in your app
app = FastAPI(title=settings.app_name)

@app.get("/info")
def get_info():
    return {
        "app": settings.app_name,
        "admin": settings.admin_email
    }

🧪 Hands-On Exercise

Ready to put your FastAPI skills to the test? Let’s build a Book Library API! 📚

Your Challenge:

Create a complete API for managing a book library with these features:

1. Book Model with title, author, ISBN, year, and available status
2. Endpoints:
   • List all books (with optional author filter)
   • Add a new book
   • Get book details by ISBN
   • Borrow a book (mark as unavailable)
   • Return a book (mark as available)
3. Validation: ISBN must be 13 digits, year must be valid
4. Background task: Log all borrow/return operations

Give it a try! Remember:

• Use Pydantic models for validation 📝
• Add proper error handling 🛡️
• Include helpful response messages 💬

from fastapi import FastAPI, HTTPException, BackgroundTasks, Query
from pydantic import BaseModel, validator
from typing import Optional, List
from datetime import datetime
import re

app = FastAPI(title="Library API 📚")

# 📖 Book model
class Book(BaseModel):
    title: str
    author: str
    isbn: str
    year: int
    available: bool = True
    
    @validator('isbn')
    def validate_isbn(cls, v):
        # Check if ISBN is 13 digits
        if not re.match(r'^\d{13}$', v):
            raise ValueError('ISBN must be exactly 13 digits! 📏')
        return v
    
    @validator('year')
    def validate_year(cls, v):
        current_year = datetime.now().year
        if v < 1000 or v > current_year:
            raise ValueError(f'Year must be between 1000 and {current_year}! 📅')
        return v

# 🗄️ In-memory database
library_db = {}

# 📝 Logging function for background
def log_transaction(action: str, isbn: str, timestamp: datetime):
    with open("library_log.txt", "a") as f:
        f.write(f"{timestamp}: {action} - ISBN: {isbn}\n")
    print(f"Logged: {action} for book {isbn} 📝")

# 📚 List all books
@app.get("/books", response_model=List[Book])
def list_books(
    author: Optional[str] = Query(None, description="Filter by author"),
    available_only: bool = Query(False, description="Show only available books")
):
    """Get all books in the library 📚"""
    books = list(library_db.values())
    
    # Apply filters
    if author:
        books = [b for b in books if author.lower() in b.author.lower()]
    if available_only:
        books = [b for b in books if b.available]
    
    return books

# ➕ Add a new book
@app.post("/books", status_code=201)
def add_book(book: Book):
    """Add a new book to the library 📖"""
    if book.isbn in library_db:
        raise HTTPException(
            status_code=400,
            detail=f"Book with ISBN {book.isbn} already exists! 🚫"
        )
    
    library_db[book.isbn] = book
    return {
        "message": f"Book '{book.title}' added successfully! 🎉",
        "isbn": book.isbn
    }

# 🔍 Get book details
@app.get("/books/{isbn}", response_model=Book)
def get_book(isbn: str):
    """Get details of a specific book 🔍"""
    if isbn not in library_db:
        raise HTTPException(
            status_code=404,
            detail=f"Book with ISBN {isbn} not found! 😔"
        )
    
    return library_db[isbn]

# 📤 Borrow a book
@app.post("/books/{isbn}/borrow")
def borrow_book(
    isbn: str,
    background_tasks: BackgroundTasks
):
    """Borrow a book from the library 📤"""
    if isbn not in library_db:
        raise HTTPException(
            status_code=404,
            detail=f"Book with ISBN {isbn} not found! 😔"
        )
    
    book = library_db[isbn]
    
    if not book.available:
        raise HTTPException(
            status_code=400,
            detail=f"Book '{book.title}' is already borrowed! 📚"
        )
    
    # Mark as borrowed
    book.available = False
    
    # Log transaction in background
    background_tasks.add_task(
        log_transaction,
        f"BORROWED: {book.title}",
        isbn,
        datetime.now()
    )
    
    return {
        "message": f"You've borrowed '{book.title}'! Enjoy reading! 📖",
        "due_date": "Please return within 14 days"
    }

# 📥 Return a book
@app.post("/books/{isbn}/return")
def return_book(
    isbn: str,
    background_tasks: BackgroundTasks
):
    """Return a borrowed book 📥"""
    if isbn not in library_db:
        raise HTTPException(
            status_code=404,
            detail=f"Book with ISBN {isbn} not found! 😔"
        )
    
    book = library_db[isbn]
    
    if book.available:
        raise HTTPException(
            status_code=400,
            detail=f"Book '{book.title}' is already in the library! 🤔"
        )
    
    # Mark as available
    book.available = True
    
    # Log transaction in background
    background_tasks.add_task(
        log_transaction,
        f"RETURNED: {book.title}",
        isbn,
        datetime.now()
    )
    
    return {
        "message": f"Thank you for returning '{book.title}'! 🙏",
        "status": "Book is now available for others"
    }

# 📊 Library statistics
@app.get("/stats")
def get_library_stats():
    """Get library statistics 📊"""
    total_books = len(library_db)
    available_books = sum(1 for book in library_db.values() if book.available)
    borrowed_books = total_books - available_books
    
    return {
        "total_books": total_books,
        "available_books": available_books,
        "borrowed_books": borrowed_books,
        "most_popular_author": max(
            set(book.author for book in library_db.values()),
            key=lambda a: sum(1 for book in library_db.values() if book.author == a)
        ) if library_db else None
    }

# 🏠 Welcome endpoint
@app.get("/")
def welcome():
    return {
        "message": "Welcome to the Library API! 📚",
        "docs": "Visit /docs for interactive documentation",
        "endpoints": [
            "GET /books - List all books",
            "POST /books - Add a new book",
            "GET /books/{isbn} - Get book details",
            "POST /books/{isbn}/borrow - Borrow a book",
            "POST /books/{isbn}/return - Return a book",
            "GET /stats - Library statistics"
        ]
    }

🎓 Key Takeaways

You’ve just mastered the fundamentals of FastAPI! Here’s what you’ve learned:

1. Automatic Validation 🛡️ - Type hints provide free validation and documentation
2. High Performance ⚡ - Built for speed with async/await support
3. Developer Friendly 🤝 - Auto-generated docs and great error messages
4. Modern Python 🐍 - Uses the latest Python features effectively
5. Production Ready 🚀 - Built-in security, testing, and deployment features

FastAPI isn’t just fast in performance - it’s fast in development too! You write less code, get more features, and build better APIs.

🤝 Next Steps

Ready to take your FastAPI skills to the next level? Here’s what to explore next:

1. Database Integration 🗄️ - Connect with SQLAlchemy or MongoDB
2. Authentication & Security 🔐 - JWT tokens, OAuth2, and permissions
3. WebSocket Support 🔌 - Real-time communication for chat apps
4. Testing Your APIs 🧪 - Unit tests and integration tests
5. Deployment 🚢 - Docker, Kubernetes, and cloud platforms

Remember, building great APIs is a journey, and you’re well on your way! Keep experimenting, keep building, and most importantly, have fun!

Happy API building! 🚀✨