📋 Task Management API

A complete REST API project demonstrating FastAPI, SQLAlchemy, JWT authentication, and Docker deployment.

🎯 Learning Objectives

After working through this project, you'll understand:

• •FastAPI Framework - Building modern Python APIs with automatic documentation
• •SQLAlchemy ORM - Database modeling with Python objects
• •JWT Authentication - Secure token-based authentication
• •Pydantic Validation - Request/response data validation with type hints
• •Dependency Injection - FastAPI's DI system for clean, testable code
• •Docker Deployment - Containerizing and deploying Python applications

Features

• •User registration and login with JWT tokens
• •Full CRUD operations for tasks
• •Task categories, priorities, and due dates
• •Filtering and pagination
• •Task completion statistics
• •PostgreSQL database (SQLite for development)
• •Docker Compose for production
• •Comprehensive test suite

🚀 Quick Start

Option 1: Local Development (Recommended for Learning)

cd 01_task_api

# Create virtual environment
python -m venv .venv
source .venv/bin/activate  # Linux/Mac
# .venv\Scripts\activate   # Windows

# Install dependencies
pip install -r requirements.txt

# Run the API
uvicorn app.main:app --reload

Option 2: Docker Compose (Production-like)

cd 01_task_api
docker-compose up -d

Access the API

• •API Root: http://localhost:8000
• •Interactive Docs (Swagger): http://localhost:8000/docs
• •Alternative Docs (ReDoc): http://localhost:8000/redoc
• •Health Check: http://localhost:8000/health

📚 API Endpoints

Authentication

Tasks (Authenticated)

Query Parameters for GET /api/tasks/

• •skip - Pagination offset (default: 0)
• •limit - Items per page (default: 100, max: 100)
• •completed - Filter by status (true/false)
• •priority - Filter by priority (low/medium/high/urgent)
• •category - Filter by category name

💡 Usage Examples

Register a User

curl -X POST http://localhost:8000/api/users/register \
  -H "Content-Type: application/json" \
  -d '{"email": "user@example.com", "username": "myuser", "password": "mypassword123"}'

Login and Get Token

curl -X POST http://localhost:8000/api/users/login \
  -d "username=myuser&password=mypassword123"

# Response: {"access_token": "eyJ...", "token_type": "bearer"}

Create a Task

curl -X POST http://localhost:8000/api/tasks/ \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "title": "Learn FastAPI",
    "description": "Complete the tutorial",
    "priority": "high",
    "category": "learning"
  }'

List Tasks with Filters

# Get high priority incomplete tasks
curl -X GET "http://localhost:8000/api/tasks/?completed=false&priority=high" \
  -H "Authorization: Bearer YOUR_TOKEN"

📁 Project Structure

01_task_api/
├── app/
│   ├── __init__.py       # Package marker
│   ├── main.py           # FastAPI application entry point
│   ├── config.py         # Settings with pydantic-settings
│   ├── database.py       # SQLAlchemy engine and session
│   ├── models.py         # ORM models (User, Task)
│   ├── schemas.py        # Pydantic request/response models
│   ├── auth.py           # JWT authentication utilities
│   └── routers/
│       ├── __init__.py
│       ├── users.py      # User registration, login, profile
│       └── tasks.py      # Task CRUD operations
├── tests/
│   ├── __init__.py
│   ├── conftest.py       # Test fixtures
│   └── test_api.py       # API tests
├── Dockerfile            # Container definition
├── docker-compose.yml    # Multi-container setup
├── requirements.txt      # Python dependencies
└── README.md

🔑 Key Concepts Explained

FastAPI Application Setup

from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Startup and shutdown events."""
    # Startup: Create database tables
    Base.metadata.create_all(bind=engine)
    yield
    # Shutdown: Cleanup if needed

app = FastAPI(
    title="Task Management API",
    lifespan=lifespan
)

Dependency Injection

from fastapi import Depends
from sqlalchemy.orm import Session

def get_db():
    """Database session dependency."""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/tasks/")
def get_tasks(
    db: Session = Depends(get_db),              # Database session
    user: User = Depends(get_current_user),     # Authenticated user
    skip: int = 0,                               # Query parameter
    limit: int = 100
):
    return db.query(Task).filter(Task.owner_id == user.id).all()

JWT Authentication Flow

# 1. User logs in with username/password
# 2. Server validates credentials
# 3. Server creates JWT token with user info
# 4. Client stores token and sends with each request
# 5. Server validates token and extracts user

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

Pydantic Schema Validation

from pydantic import BaseModel, Field, EmailStr

class TaskCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    description: str | None = Field(None, max_length=1000)
    priority: TaskPriority = TaskPriority.MEDIUM

# Automatic validation on request:
# - title is required (...)
# - title length between 1-200 chars
# - description is optional, max 1000 chars
# - priority defaults to MEDIUM

🧪 Running Tests

# Install test dependencies
pip install pytest pytest-asyncio httpx

# Run all tests
pytest tests/ -v

# Run with coverage
pip install pytest-cov
pytest tests/ --cov=app --cov-report=html

🔧 Configuration

Environment variables (use .env file):

# Database (SQLite by default, PostgreSQL for production)
DATABASE_URL=sqlite:///./tasks.db
# DATABASE_URL=postgresql://user:pass@localhost:5432/tasks

# JWT Settings
SECRET_KEY=your-super-secret-key-change-in-production
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30

# App Settings
DEBUG=true

📈 Extending the API

Add a New Model

1. •Create model in models.py:

class Category(Base):
    __tablename__ = "categories"
    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True)

1. •Add schemas in schemas.py
2. •Create router in routers/categories.py
3. •Register router in main.py

Add Pagination Response

from math import ceil

class PaginatedTasks(BaseModel):
    items: list[TaskResponse]
    total: int
    page: int
    per_page: int
    pages: int

@router.get("/", response_model=PaginatedTasks)
def list_tasks(page: int = 1, per_page: int = 20):
    total = db.query(Task).count()
    items = db.query(Task).offset((page-1)*per_page).limit(per_page).all()
    return {
        "items": items,
        "total": total,
        "page": page,
        "per_page": per_page,
        "pages": ceil(total / per_page)
    }

📚 Related Learning

• •FastAPI Official Tutorial
• •SQLAlchemy ORM Tutorial
• •JWT.io - Learn about JSON Web Tokens
• •Pydantic Documentation

✅ Project Checklist

• • Run the API and explore /docs
• • Register a user and login
• • Create, update, and delete tasks
• • Try filtering tasks by priority and status
• • Run the test suite
• • Add a new field to tasks (e.g., tags)
• • Deploy with Docker Compose
• • Add rate limiting middleware