FastAPI Complete Course Module 17: Real-World Projects

Introduction

Welcome to Module 17, where theory meets practice. In this chapter, you will learn how to build five complete, real-world FastAPI projects. Each project is designed to teach you a different aspect of FastAPI development—from CRUD operations and authentication to AI integration and complex business logic. By the end of this module, you will have a portfolio of working APIs that demonstrate your ability to handle real client requirements.

The primary focus keyword for this chapter is FastAPI Real-World Projects. We will explore project architecture, reusable class and service patterns, and provide practical endpoint examples for each idea. Let’s start building.

1. Student Management System API

Project Overview

A Student Management System (SMS) API allows schools to manage student records, courses, enrollments, and grades. This is a classic CRUD (Create, Read, Update, Delete) application with relationships between tables.

Project Architecture

• Models: Student, Course, Enrollment, Grade
• Schemas: Pydantic models for request/response validation
• Services: Business logic layer (StudentService, EnrollmentService)
• Routers: API endpoints grouped by resource
• Database: SQLite (development) / PostgreSQL (production)

Reusable Service Pattern

We use a generic BaseService class to avoid repeating CRUD logic for every model.

from sqlalchemy.orm import Session
from typing import TypeVar, Generic, Type, List, Optional
from pydantic import BaseModel

ModelType = TypeVar("ModelType")
CreateSchemaType = TypeVar("CreateSchemaType")
UpdateSchemaType = TypeVar("UpdateSchemaType")

class BaseService(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
    def __init__(self, model: Type[ModelType], db: Session):
        self.model = model
        self.db = db

    def create(self, schema: CreateSchemaType) -> ModelType:
        db_obj = self.model(**schema.dict())
        self.db.add(db_obj)
        self.db.commit()
        self.db.refresh(db_obj)
        return db_obj

    def get(self, id: int) -> Optional[ModelType]:
        return self.db.query(self.model).filter(self.model.id == id).first()

    def get_all(self, skip: int = 0, limit: int = 100) -> List[ModelType]:
        return self.db.query(self.model).offset(skip).limit(limit).all()

    def update(self, id: int, schema: UpdateSchemaType) -> Optional[ModelType]:
        db_obj = self.get(id)
        if db_obj:
            for key, value in schema.dict(exclude_unset=True).items():
                setattr(db_obj, key, value)
            self.db.commit()
            self.db.refresh(db_obj)
        return db_obj

    def delete(self, id: int) -> bool:
        db_obj = self.get(id)
        if db_obj:
            self.db.delete(db_obj)
            self.db.commit()
            return True
        return False

Explanation: This generic service uses Python generics to work with any SQLAlchemy model. The create method takes a Pydantic schema, converts it to a dictionary, and creates a database object. get and get_all handle retrieval with optional pagination. update only updates fields that are provided (using exclude_unset=True). delete returns a boolean for easy error handling.

Practical Endpoint Example

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app import schemas, models, services
from app.database import get_db

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

@router.post("/", response_model=schemas.StudentResponse)
def create_student(student: schemas.StudentCreate, db: Session = Depends(get_db)):
    service = services.StudentService(model=models.Student, db=db)
    return service.create(student)

@router.get("/{student_id}", response_model=schemas.StudentResponse)
def get_student(student_id: int, db: Session = Depends(get_db)):
    service = services.StudentService(model=models.Student, db=db)
    db_student = service.get(student_id)
    if not db_student:
        raise HTTPException(status_code=404, detail="Student not found")
    return db_student

@router.get("/", response_model=list[schemas.StudentResponse])
def list_students(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    service = services.StudentService(model=models.Student, db=db)
    return service.get_all(skip=skip, limit=limit)

Explanation: Each endpoint instantiates the StudentService with the model and database session. The response_model ensures only safe fields are returned. The get_student endpoint returns a 404 if the student doesn’t exist.

2. E-Commerce Backend API

Project Overview

An e-commerce backend handles products, categories, shopping carts, orders, and payments. This project introduces authentication, authorization, and complex relationships.

Project Architecture

• Models: User, Product, Category, Cart, CartItem, Order, OrderItem
• Authentication: JWT tokens with OAuth2 password flow
• Services: AuthService, ProductService, CartService, OrderService
• Middleware: Rate limiting, CORS

Reusable Class Pattern for Authentication

from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

class AuthService:
    def __init__(self, secret_key: str, algorithm: str = "HS256", expire_minutes: int = 30):
        self.secret_key = secret_key
        self.algorithm = algorithm
        self.expire_minutes = expire_minutes

    def hash_password(self, password: str) -> str:
        return pwd_context.hash(password)

    def verify_password(self, plain_password: str, hashed_password: str) -> bool:
        return pwd_context.verify(plain_password, hashed_password)

    def create_access_token(self, data: dict) -> str:
        to_encode = data.copy()
        expire = datetime.utcnow() + timedelta(minutes=self.expire_minutes)
        to_encode.update({"exp": expire})
        return jwt.encode(to_encode, self.secret_key, algorithm=self.algorithm)

    def decode_token(self, token: str) -> dict:
        try:
            payload = jwt.decode(token, self.secret_key, algorithms=[self.algorithm])
            return payload
        except JWTError:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid authentication credentials",
            )

Explanation: This AuthService centralizes all authentication logic. hash_password uses bcrypt for secure password storage. create_access_token generates a JWT with an expiration time. decode_token validates the token and raises a 401 error if invalid.

Practical Endpoint: Add to Cart

@router.post("/cart/add", response_model=schemas.CartResponse)
def add_to_cart(
    item: schemas.CartItemCreate,
    db: Session = Depends(get_db),
    current_user: models.User = Depends(get_current_user)
):
    cart_service = services.CartService(db)
    return cart_service.add_item(user_id=current_user.id, product_id=item.product_id, quantity=item.quantity)

Explanation: The endpoint requires authentication (get_current_user). The CartService.add_item method handles creating or updating cart items. This pattern keeps the endpoint thin and the business logic testable.

3. AI Chatbot Backend API

Project Overview

An AI Chatbot backend integrates with OpenAI or Hugging Face models, manages conversation history, and provides context-aware responses. This project teaches asynchronous programming and third-party API integration.

Project Architecture

• Models: Conversation, Message
• Services: LLMService (OpenAI), ConversationService
• Async: Use httpx.AsyncClient for non-blocking requests
• Streaming: Server-Sent Events (SSE) for real-time responses

Reusable LLM Service

import httpx
from typing import List, Dict, Optional
from app.config import settings

class LLMService:
    def __init__(self, api_key: str = settings.OPENAI_API_KEY, model: str = "gpt-3.5-turbo"):
        self.api_key = api_key
        self.model = model
        self.base_url = "https://api.openai.com/v1/chat/completions"

    async def generate_response(self, messages: List[Dict[str, str]]) -> str:
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        payload = {
            "model": self.model,
            "messages": messages,
            "temperature": 0.7
        }
        async with httpx.AsyncClient() as client:
            response = await client.post(self.base_url, json=payload, headers=headers)
            response.raise_for_status()
            return response.json()["choices"][0]["message"]["content"]

    async def stream_response(self, messages: List[Dict[str, str]]):
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        payload = {
            "model": self.model,
            "messages": messages,
            "stream": True
        }
        async with httpx.AsyncClient() as client:
            async with client.stream("POST", self.base_url, json=payload, headers=headers) as response:
                async for line in response.aiter_lines():
                    if line.startswith("data: "):
                        data = line[6:]
                        if data != "[DONE]":
                            yield data

Explanation: The service uses httpx.AsyncClient for non-blocking HTTP calls. generate_response returns the full response. stream_response is a generator that yields data chunks for real-time streaming. The settings.OPENAI_API_KEY comes from environment variables.

Practical Endpoint: Chat with History

@router.post("/chat")
async def chat(
    request: schemas.ChatRequest,
    db: Session = Depends(get_db),
    current_user: models.User = Depends(get_current_user)
):
    conversation_service = services.ConversationService(db)
    llm_service = services.LLMService()
    
    # Get or create conversation
    conversation = conversation_service.get_or_create(current_user.id, request.conversation_id)
    
    # Build message history
    messages = conversation_service.get_history(conversation.id)
    messages.append({"role": "user", "content": request.message})
    
    # Get AI response
    ai_response = await llm_service.generate_response(messages)
    
    # Save both messages
    conversation_service.add_message(conversation.id, "user", request.message)
    conversation_service.add_message(conversation.id, "assistant", ai_response)
    
    return {"response": ai_response, "conversation_id": conversation.id}

Explanation: The endpoint manages conversation context. It retrieves or creates a conversation, builds the message history, calls the LLM, and saves both user and assistant messages. The conversation_id allows the frontend to continue the same conversation.

4. Hospital Management API

Project Overview

A Hospital Management API handles patients, doctors, appointments, medical records, and billing. This project introduces role-based access control (RBAC), scheduling logic, and sensitive data handling.

Project Architecture

• Models: Patient, Doctor, Appointment, MedicalRecord, Invoice
• Roles: admin, doctor, nurse, patient
• Services: AppointmentService, BillingService, MedicalRecordService
• Validation: Time slot availability, duplicate prevention

Reusable RBAC Dependency

from fastapi import Depends, HTTPException, status
from enum import Enum

class UserRole(str, Enum):
    ADMIN = "admin"
    DOCTOR = "doctor"
    NURSE = "nurse"
    PATIENT = "patient"

def require_role(required_role: UserRole):
    def role_checker(current_user: models.User = Depends(get_current_user)):
        if current_user.role != required_role:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Role {required_role} required"
            )
        return current_user
    return role_checker

# Usage in router
@router.get("/patients/{patient_id}/records", dependencies=[Depends(require_role(UserRole.DOCTOR))])
def get_patient_records(patient_id: int, db: Session = Depends(get_db)):
    # Only doctors can access this
    pass

Explanation: The require_role function returns a dependency that checks the user’s role. This is a reusable pattern that can be applied to any endpoint. The dependencies parameter in the router decorator applies the check automatically.

Practical Endpoint: Book Appointment with Conflict Detection

@router.post("/appointments", response_model=schemas.AppointmentResponse)
def book_appointment(
    appointment: schemas.AppointmentCreate,
    db: Session = Depends(get_db),
    current_user: models.User = Depends(get_current_user)
):
    appointment_service = services.AppointmentService(db)
    
    # Check for time conflicts
    if appointment_service.has_conflict(appointment.doctor_id, appointment.start_time, appointment.end_time):
        raise HTTPException(status_code=409, detail="Time slot already booked")
    
    # Check doctor availability
    if not appointment_service.is_doctor_available(appointment.doctor_id, appointment.start_time):
        raise HTTPException(status_code=400, detail="Doctor not available at this time")
    
    return appointment_service.create(appointment)

Explanation: Before creating an appointment, the service checks for conflicts and doctor availability. The has_conflict method queries the database for overlapping appointments. This prevents double-booking and ensures data integrity.

5. School AI Assistant API

Project Overview

A School AI Assistant API combines student management with AI capabilities. It can answer student questions, generate study plans, grade assignments, and provide personalized learning recommendations.

Project Architecture

• Models: Student, Course, Assignment, Submission, Feedback, StudyPlan
• AI Integration: Use LLM for grading and recommendations
• Services: GradingService, StudyPlanService, FeedbackService
• Background Tasks: Process assignments asynchronously

Reusable Background Task Pattern

from fastapi import BackgroundTasks
from sqlalchemy.orm import Session

class GradingService:
    def __init__(self, db: Session):
        self.db = db
        self.llm = services.LLMService()

    async def grade_submission_background(self, submission_id: int):
        submission = self.db.query(models.Submission).filter(models.Submission.id == submission_id).first()
        if not submission:
            return
        
        # Create grading prompt
        prompt = f"Grade this {submission.assignment.subject} assignment. Score out of 100:n{submission.content}"
        
        # Get AI grade
        result = await self.llm.generate_response([{"role": "user", "content": prompt}])
        
        # Parse and save grade
        grade = int(result.split("/")[0])  # Simplified parsing
        feedback = models.Feedback(
            submission_id=submission_id,
            grade=grade,
            comments=result,
            graded_by="AI"
        )
        self.db.add(feedback)
        self.db.commit()

# Endpoint using background task
@router.post("/submissions/{submission_id}/grade")
async def grade_submission(
    submission_id: int,
    background_tasks: BackgroundTasks,
    db: Session = Depends(get_db)
):
    grading_service = services.GradingService(db)
    background_tasks.add_task(grading_service.grade_submission_background, submission_id)
    return {"message": "Grading started", "submission_id": submission_id}

Explanation: Grading is CPU-intensive and time-consuming. Using BackgroundTasks, the endpoint returns immediately while grading happens in the background. The grade_submission_background method queries the submission, builds a prompt, calls the LLM, and saves the result.

Practical Endpoint: Generate Study Plan

@router.post("/study-plans/generate", response_model=schemas.StudyPlanResponse)
async def generate_study_plan(
    request: schemas.StudyPlanRequest,
    db: Session = Depends(get_db),
    current_user: models.User = Depends(get_current_user)
):
    study_plan_service = services.StudyPlanService(db)
    llm_service = services.LLMService()
    
    # Get student's weak areas from past assignments
    weak_areas = study_plan_service.get_weak_areas(current_user.id)
    
    # Generate personalized plan
    prompt = f"Create a 1-week study plan for a student weak in {', '.join(weak_areas)}. Include daily topics and practice exercises."
    plan_text = await llm_service.generate_response([{"role": "user", "content": prompt}])
    
    # Save plan
    study_plan = study_plan_service.create(
        user_id=current_user.id,
        plan_text=plan_text,
        weak_areas=weak_areas
    )
    return study_plan

Explanation: The endpoint analyzes the student’s performance data to identify weak areas, then uses AI to generate a personalized study plan. The plan is saved for future reference.

Common Mistakes

• Not using dependency injection properly: Always use Depends() for database sessions and authentication. Avoid creating services inside endpoints without proper injection.
• Exposing sensitive data: Never return password hashes or internal IDs in API responses. Use Pydantic response models to control what is exposed.
• Ignoring async for I/O operations: Database queries and HTTP calls should be async. Use async def for endpoints that await external services.
• Hardcoding configuration: Store API keys, database URLs, and secrets in environment variables or a .env file. Use Pydantic Settings for validation.
• Not handling background task errors: Background tasks run silently. Implement logging and error tracking for background operations.

Practice Task

Build a Task Management API that combines patterns from all five projects:

1. Create models: User, Project, Task, Comment
2. Implement JWT authentication (like E-Commerce)
3. Use generic CRUD service (like Student Management)
4. Add role-based access (Admin, Manager, Member) (like Hospital)
5. Integrate AI for task prioritization (like School AI Assistant)
6. Use background tasks for sending email notifications (like Grading)

Expected outcome: A fully functional API with at least 10 endpoints, authentication, role control, and one AI-powered feature.

Summary

In this chapter, you learned how to build five real-world FastAPI projects:

• Student Management System: Generic CRUD services and database relationships
• E-Commerce Backend: JWT authentication and cart management
• AI Chatbot Backend: Async LLM integration and conversation management
• Hospital Management: Role-based access control and conflict detection
• School AI Assistant: Background tasks and personalized AI features

Each project introduced reusable patterns—generic services, authentication classes, RBAC dependencies, and background task handling—that you can apply to any FastAPI application. You now have a solid foundation for building production-ready APIs.

FAQs

Q1: Do I need to use all these patterns in every project?

No. Choose patterns based on your project’s complexity. A simple CRUD API may not need RBAC or background tasks. Start simple and add complexity as needed.

Q2: How do I handle database migrations for these projects?

Use Alembic with SQLAlchemy. After defining your models, run alembic init alembic, configure alembic.ini with your database URL, then use alembic revision --autogenerate -m "message" and alembic upgrade head.

Q3: Can I use MongoDB instead of PostgreSQL?

Yes. FastAPI works with any database. Use Beanie or Motor for async MongoDB. The service pattern remains the same; only the database queries change.

Q4: How do I deploy these APIs?

Use Docker to containerize your application. Deploy to cloud platforms like AWS (ECS), Google Cloud Run, or DigitalOcean App Platform. Use environment variables for configuration.

Q5: How do I test these APIs?

Use FastAPI’s TestClient for unit tests. For integration tests, use a test database. Write tests for each service method and endpoint. Use pytest fixtures for database sessions.

You are now ready for the capstone project. You have learned to architect, build, and deploy real-world FastAPI applications. The patterns and techniques from this module will serve as your toolkit for any API development challenge. Go build something amazing.

More Practical Examples

Let’s expand your project toolkit with two additional real-world FastAPI applications that demonstrate common patterns you’ll encounter frequently.

Task Management API with Background Tasks

Many applications need to run operations after returning a response—sending emails, processing files, or updating caches. FastAPI’s BackgroundTasks makes this trivial.

from fastapi import FastAPI, BackgroundTasks, HTTPException
from pydantic import BaseModel
import time

app = FastAPI(title="Task Manager with Background Jobs")

class TaskCreate(BaseModel):
    title: str
    description: str = ""
    priority: int = 1

tasks_db = []

def send_notification(task_title: str):
    """Simulate sending an email notification after task creation."""
    time.sleep(2)  # Simulate network delay
    print(f"Notification sent for task: {task_title}")

@app.post("/tasks/", status_code=201)
async def create_task(task: TaskCreate, background_tasks: BackgroundTasks):
    task_id = len(tasks_db) + 1
    new_task = {
        "id": task_id,
        "title": task.title,
        "description": task.description,
        "priority": task.priority,
        "completed": False
    }
    tasks_db.append(new_task)
    
    # Schedule notification in background
    background_tasks.add_task(send_notification, task.title)
    
    return {"message": "Task created", "task": new_task}

@app.get("/tasks/")
async def list_tasks():
    return {"tasks": tasks_db}

@app.get("/tasks/{task_id}")
async def get_task(task_id: int):
    if task_id  len(tasks_db):
        raise HTTPException(status_code=404, detail="Task not found")
    return tasks_db[task_id - 1]

Explanation: This example shows how to handle non-blocking operations. When a task is created, the endpoint immediately returns the response while FastAPI runs send_notification in the background. The BackgroundTasks parameter is automatically injected by FastAPI. This pattern is essential for real-world APIs where you need to keep response times low while performing secondary work.

URL Shortener API with Redirect

URL shorteners are a classic project that teaches you about redirects, hashing, and database lookups.

import hashlib
import string
from fastapi import FastAPI, HTTPException, Response
from pydantic import BaseModel

app = FastAPI(title="URL Shortener")

# In-memory storage (use a real DB in production)
url_map = {}

class URLRequest(BaseModel):
    url: str

def generate_short_code(url: str) -> str:
    """Create a 6-character short code from URL hash."""
    hash_object = hashlib.md5(url.encode())
    hex_digest = hash_object.hexdigest()
    # Take first 6 hex characters
    return hex_digest[:6]

@app.post("/shorten")
async def shorten_url(request: URLRequest):
    # Validate URL format (basic check)
    if not request.url.startswith(("http://", "https://")):
        raise HTTPException(status_code=400, detail="Invalid URL format")
    
    short_code = generate_short_code(request.url)
    
    # Handle collisions (simple: append counter)
    original_code = short_code
    counter = 0
    while short_code in url_map and url_map[short_code] != request.url:
        counter += 1
        short_code = original_code + str(counter)
    
    url_map[short_code] = request.url
    return {"short_url": f"http://localhost:8000/{short_code}", "original_url": request.url}

@app.get("/{short_code}")
async def redirect_to_url(short_code: str):
    if short_code not in url_map:
        raise HTTPException(status_code=404, detail="Short URL not found")
    # Return a redirect response (HTTP 307 preserves HTTP method)
    return Response(status_code=307, headers={"Location": url_map[short_code]})

Explanation: This API accepts a long URL and returns a shortened version. The generate_short_code function uses MD5 hashing to create a unique identifier. The redirect endpoint uses HTTP 307 to send the client to the original URL. In production, you’d replace the in-memory dictionary with a database like Redis for persistence and speed.

Class-Based Example

While FastAPI works beautifully with functions, using classes can help organize related endpoints and share state. Here’s a complete example of a Book Inventory API using a class-based approach with dependency injection.

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI(title="Book Inventory (Class-Based)")

# Pydantic models
class Book(BaseModel):
    isbn: str
    title: str
    author: str
    price: float
    quantity: int

class BookUpdate(BaseModel):
    title: Optional[str] = None
    author: Optional[str] = None
    price: Optional[float] = None
    quantity: Optional[int] = None

# In-memory storage class
class BookRepository:
    def __init__(self):
        self._books = {}
    
    def add_book(self, book: Book) -> Book:
        if book.isbn in self._books:
            raise HTTPException(status_code=400, detail="Book already exists")
        self._books[book.isbn] = book
        return book
    
    def get_book(self, isbn: str) -> Book:
        if isbn not in self._books:
            raise HTTPException(status_code=404, detail="Book not found")
        return self._books[isbn]
    
    def get_all_books(self) -> List[Book]:
        return list(self._books.values())
    
    def update_book(self, isbn: str, updates: BookUpdate) -> Book:
        if isbn not in self._books:
            raise HTTPException(status_code=404, detail="Book not found")
        book = self._books[isbn]
        update_data = updates.dict(exclude_unset=True)
        for field, value in update_data.items():
            setattr(book, field, value)
        self._books[isbn] = book
        return book
    
    def delete_book(self, isbn: str) -> dict:
        if isbn not in self._books:
            raise HTTPException(status_code=404, detail="Book not found")
        del self._books[isbn]
        return {"message": "Book deleted"}

# Create singleton instance
repo = BookRepository()

# Dependency to inject the repository
def get_repository() -> BookRepository:
    return repo

# Endpoints using dependency injection
@app.post("/books/", response_model=Book)
async def create_book(book: Book, repo: BookRepository = Depends(get_repository)):
    return repo.add_book(book)

@app.get("/books/", response_model=List[Book])
async def list_books(repo: BookRepository = Depends(get_repository)):
    return repo.get_all_books()

@app.get("/books/{isbn}", response_model=Book)
async def get_book(isbn: str, repo: BookRepository = Depends(get_repository)):
    return repo.get_book(isbn)

@app.put("/books/{isbn}", response_model=Book)
async def update_book(isbn: str, updates: BookUpdate, repo: BookRepository = Depends(get_repository)):
    return repo.update_book(isbn, updates)

@app.delete("/books/{isbn}")
async def delete_book(isbn: str, repo: BookRepository = Depends(get_repository)):
    return repo.delete_book(isbn)

Explanation: This class-based design separates concerns cleanly:

• BookRepository encapsulates all data access logic, making it easy to swap the in-memory store for a database later.
• Dependency injection (Depends(get_repository)) allows FastAPI to provide the repository to each endpoint. This makes testing easier—you can inject a mock repository.
• The BookUpdate model with Optional fields enables partial updates via PUT, using exclude_unset=True to only update provided fields.

This pattern scales well for larger applications. You can extend BookRepository with methods like search_by_author() or get_low_stock_books() without changing your endpoint code.

Step-by-Step Exercise

Now it’s your turn. Build a Simple Blog API with the following requirements. This exercise will reinforce everything you’ve learned.

Requirements

1. Create a FastAPI app with two models: Post (id, title, content, author, created_at) and Comment (id, post_id, author, content, created_at).
2. Implement CRUD endpoints for posts (create, read all, read one, update, delete).
3. Implement endpoints to add a comment to a post and list comments for a post.
4. Add validation: title must be at least 5 characters, content at least 20 characters.
5. Use BackgroundTasks to log “New post created” after each post creation.
6. Add a search endpoint /posts/search/?q=term that returns posts where title or content contains the term (case-insensitive).

Starter Code

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

app = FastAPI(title="Blog API Exercise")

# Your code here
# 1. Define Post and Comment models
# 2. Create in-memory storage (lists or dicts)
# 3. Implement all required endpoints

Solution (Partial – Endpoints Only)

# Continuing from starter code...

posts_db = {}
comments_db = {}
post_counter = 0
comment_counter = 0

class PostCreate(BaseModel):
    title: str
    content: str
    author: str

    @validator("title")
    def title_min_length(cls, v):
        if len(v) < 5:
            raise ValueError("Title must be at least 5 characters")
        return v

    @validator("content")
    def content_min_length(cls, v):
        if len(v) < 20:
            raise ValueError("Content must be at least 20 characters")
        return v

class Post(PostCreate):
    id: int
    created_at: datetime

class CommentCreate(BaseModel):
    author: str
    content: str

class Comment(CommentCreate):
    id: int
    post_id: int
    created_at: datetime

def log_post_creation(post_id: int):
    print(f"[LOG] New post created with ID: {post_id} at {datetime.now()}")

@app.post("/posts/", response_model=Post, status_code=201)
async def create_post(post: PostCreate, background_tasks: BackgroundTasks):
    global post_counter
    post_counter += 1
    new_post = Post(
        id=post_counter,
        **post.dict(),
        created_at=datetime.now()
    )
    posts_db[post_counter] = new_post
    background_tasks.add_task(log_post_creation, post_counter)
    return new_post

@app.get("/posts/", response_model=List[Post])
async def list_posts():
    return list(posts_db.values())

@app.get("/posts/{post_id}", response_model=Post)
async def get_post(post_id: int):
    if post_id not in posts_db:
        raise HTTPException(status_code=404, detail="Post not found")
    return posts_db[post_id]

@app.put("/posts/{post_id}", response_model=Post)
async def update_post(post_id: int, post: PostCreate):
    if post_id not in posts_db:
        raise HTTPException(status_code=404, detail="Post not found")
    updated_post = Post(
        id=post_id,
        **post.dict(),
        created_at=posts_db[post_id].created_at
    )
    posts_db[post_id] = updated_post
    return updated_post

@app.delete("/posts/{post_id}")
async def delete_post(post_id: int):
    if post_id not in posts_db:
        raise HTTPException(status_code=404, detail="Post not found")
    del posts_db[post_id]
    # Also delete associated comments
    comments_to_remove = [cid for cid, c in comments_db.items() if c.post_id == post_id]
    for cid in comments_to_remove:
        del comments_db[cid]
    return {"message": "Post and associated comments deleted"}

@app.post("/posts/{post_id}/comments/", response_model=Comment, status_code=201)
async def add_comment(post_id: int, comment: CommentCreate):
    if post_id not in posts_db:
        raise HTTPException(status_code=404, detail="Post not found")
    global comment_counter
    comment_counter += 1
    new_comment = Comment(
        id=comment_counter,
        post_id=post_id,
        **comment.dict(),
        created_at=datetime.now()
    )
    comments_db[comment_counter] = new_comment
    return new_comment

@app.get("/posts/{post_id}/comments/", response_model=List[Comment])
async def list_comments(post_id: int):
    if post_id not in posts_db:
        raise HTTPException(status_code=404, detail="Post not found")
    return [c for c in comments_db.values() if c.post_id == post_id]

@app.get("/posts/search/", response_model=List[Post])
async def search_posts(q: str = Query(..., min_length=2)):
    results = []
    for post in posts_db.values():
        if q.lower() in post.title.lower() or q.lower() in post.content.lower():
            results.append(post)
    return results

Explanation: This exercise combines everything: models with validation, CRUD operations, background tasks, search functionality, and nested resources (comments under posts). Try running it with uvicorn main:app --reload and test with the interactive docs at /docs.

Interview and Job Use Cases

Understanding these patterns will help you in technical interviews and on the job. Here are common scenarios where FastAPI skills shine.

Interview Questions You Might Face

1. “How do you handle database connections in FastAPI?” — Use dependency injection with Depends() to create and close connections per request. Example: def get_db(): db = SessionLocal(); yield db; db.close()
2. “Explain middleware in FastAPI.” — Middleware processes every request/response. Common uses: CORS, authentication, request logging. You can create custom middleware using @app.middleware("http").
3. “How do you implement pagination?” — Accept skip and limit query parameters. Example: def list_items(skip: int = 0, limit: int = 10): return db[skip: skip+limit]
4. “What’s the difference between Query, Path, and Body parameters?” — Query: from URL query string. Path: from URL path. Body: from request body (JSON). FastAPI validates all automatically.
5. “How do you handle file uploads?” — Use UploadFile from fastapi and File from starlette. Example: async def upload(file: UploadFile = File(...))

Real-World Job Tasks

• Building a microservice for user authentication — Implement JWT tokens, password hashing (bcrypt), and OAuth2 flows.
• Creating an API gateway — Route requests to multiple backend services, add rate limiting, and aggregate responses.
• Developing a real-time dashboard — Use WebSocket endpoints in FastAPI to push live data to frontend clients.
• Automating deployment — Write a CI/CD pipeline that runs your FastAPI tests, builds a Docker image, and deploys to Kubernetes.
• Integrating with external APIs — Use httpx (async HTTP client) inside your FastAPI endpoints to call third-party services.

Deployment Configuration Example

Here’s a typical Dockerfile and docker-compose.yml for a FastAPI app:

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
version: '3.8'
services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://user:pass@db:5432/mydb
    depends_on:
      - db
  db:
    image: postgres:15
    environment:
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=pass
      - POSTGRES_DB=mydb

Explanation: This setup containerizes your FastAPI app with PostgreSQL. The DATABASE_URL environment variable lets you switch between local development and production databases without code changes.

Extra Beginner FAQs

Q: What’s the difference between FastAPI and Flask?

A: FastAPI is async-first, auto-generates OpenAPI docs, and has built-in validation via Pydantic. Flask is synchronous and requires more manual setup for validation and docs. FastAPI is generally faster for I/O-bound tasks.

Q: Do I need to use async/await everywhere?

A: No. FastAPI works fine with synchronous code. Use def instead of async def for endpoints that perform CPU-bound work or use blocking libraries (like some ORMs). FastAPI runs synchronous functions in a thread pool.

Q: How do I connect to a real database?

A: Use an ORM like SQLAlchemy (with async support via asyncpg) or Tortoise-ORM. Install the database driver (e.g., pip install asyncpg), create a session dependency, and inject it into your endpoints.

Q: Why does my POST request return 422 Unprocessable Entity?

A: FastAPI validates request bodies against your Pydantic model. Check that your JSON matches the model’s field types and required fields. The error response includes details about which field failed.

Q: How do I handle CORS for frontend development?

A: Import CORSMiddleware and add it to your app: app.add_middleware(CORSMiddleware, allow_origins=["http://localhost:3000"], allow_methods=["*"], allow_headers=["*"]).

Q: Can I use FastAPI with Django?

A: Yes! You can run FastAPI alongside Django using Django’s ASGI support. Use FastAPI for high-performance API endpoints and Django for admin, auth, and ORM. This is common in large projects.

Q: How do I add authentication?

A: FastAPI supports OAuth2 with JWT out of the box. Use OAuth2PasswordBearer for token extraction, python-jose for JWT encoding/decoding, and passlib for password hashing. See FastAPI’s official security tutorial.

Q: What’s the best way to structure a large FastAPI project?

A: Use a modular structure with separate files for models, schemas, routers, services, and dependencies. Example: app/ with subfolders routers/, models/, schemas/, services/, core/ (config, security).

Q: How do I test my FastAPI endpoints?

A: Use FastAPI’s TestClient (based on httpx). Write tests in test_main.py that make requests and assert responses. Example: client = TestClient(app); response = client.get("/items/"); assert response.status_code == 200.

Q: Can I use WebSockets with FastAPI?

A: Yes! FastAPI supports WebSockets natively. Define a WebSocket endpoint using @app.websocket("/ws") and use await websocket.receive_text() / await websocket.send_text() for real-time communication.