Ravi Gupta

Posted on Mar 2

Building a Production-Ready Task Management API with FastAPI: Development & Implementation (Part 2)

#fastapi #api #backend #python

Building a Production-Ready Task Management API with FastAPI: Development & Implementation (Part 2)

Part 2 of 3: Architecture is comfortable. Implementation is humbling.

In Part 1, everything looked clean.

The layers were well separated.

The database schema made sense.

The architecture diagram looked impressive.

And then I started writing code.
That’s when things stopped being theoretical.
Not just what worked — but what broke, what surprised me, and that forced me to level up.
This article covers JWT authentication, CRUD operations, advanced features, and the real challenges of building production-ready APIs.

Quick Links:

Reading time: ~12 minutes

🎯 Introduction & Recap

In Part 1, I designed the architecture for a production-ready Task Management API using FastAPI and PostgreSQL.

What we covered:

Technology stack selection (FastAPI, PostgreSQL, SQLAlchemy 2.0)
Database schema design (Users, Tasks, Categories)
Clean architecture layers (API → Service → Repository → Database)
Project structure organization

Phase 1 Result: A solid foundation ready for implementation.
On paper, it was solid.

But architecture doesn’t test your assumptions, Implementation does.

Phase 2 Goal: Build the actual features while maintaining architectural integrity.
And discovering that “working” is not the same as “production-ready.”

This article covers the development journey - the features I built, patterns I implemented, and challenges I faced turning architecture diagrams into working code.

🔐 JWT Authentication Implementation

The Challenge

Coming from Spring Boot, I was used to this:

@Configuration
@EnableWebSecurity
public class SecurityConfig {
    // Spring Security handles everything
}

In FastAPI, there's no magic @EnableWebSecurity.

You build authentication from the ground up.

What I Built

Complete authentication system with:

User Registration
- Email validation
- Password strength requirements
- Duplicate email/username checking
- Automatic password hashing
Login Flow
- Credential verification
- Access token generation (15 min expiry)
- Refresh token generation (7 days expiry)
- Token storage strategy
Token Management
- Token validation middleware
- Token refresh endpoint
- Automatic token expiry handling
- Logout mechanism
Security Features
- Email verification flow
- Password reset mechanism
- Account activation
- Rate limiting on auth endpoints

Implementation Deep Dive

1. Password Hashing with bcrypt

# app/core/security.py
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str) -> str:
    """Hash a password using bcrypt."""
    return pwd_context.hash(password)

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """Verify a password against a hashed password."""
    return pwd_context.verify(plain_password, hashed_password)

Why bcrypt?

Industry standard for password hashing
Built-in salt generation
Adaptive cost factor (future-proof against hardware improvements)
Slow by design (prevents brute force)

This was my first mindset shift:

FastAPI gives flexibility.
Security becomes your responsibility.

Spring Boot comparison: Similar to BCryptPasswordEncoder, but more explicit configuration.

2. JWT Token Generation

# app/core/security.py
from datetime import datetime, timedelta
from jose import jwt
from app.config import settings

def create_access_token(data: dict, expires_delta: timedelta = None) -> str:
    """Create JWT access token."""
    to_encode = data.copy()

    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)

    to_encode.update({"exp": expire, "type": "access"})
    encoded_jwt = jwt.encode(
        to_encode, 
        settings.SECRET_KEY, 
        algorithm=settings.ALGORITHM
    )
    return encoded_jwt

def create_refresh_token(data: dict) -> str:
    """Create JWT refresh token."""
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(days=7)
    to_encode.update({"exp": expire, "type": "refresh"})

    encoded_jwt = jwt.encode(
        to_encode,
        settings.REFRESH_SECRET_KEY,
        algorithm=settings.ALGORITHM
    )
    return encoded_jwt

Design decisions here matter.

JWT is just a token format.

Security comes from validation.

Key decisions:

Two token types:
- Access tokens (short-lived, 15 min)
- Refresh tokens (long-lived, 7 days)
Separate secrets:
- Different keys for access vs refresh tokens
- Compromised access token ≠ compromised refresh token
Token payload:

   {
     "sub": "user_id",
     "exp": 1234567890,
     "type": "access"
   }

3. Token Validation Dependency

How it works:

Extract JWT from Authorization: Bearer <token> header
Decode and validate token signature
Check token type (must be "access")
Fetch user from database
Verify user is active
Return user object

Usage in routes:

@router.get("/me", response_model=UserResponse)
async def get_current_user_profile(
    current_user: User = Depends(get_current_user)
):
    """Get current user's profile."""
    return current_user

Clean. Type-safe. Reusable.

4. Login Endpoint Implementation

# app/api/v1/auth.py
from fastapi import APIRouter, Depends, HTTPException, status
from app.schemas.token import TokenResponse
from app.schemas.user import UserLogin

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

@router.post("/login", response_model=TokenResponse)
async def login(
    credentials: UserLogin,
    db: AsyncSession = Depends(get_db)
):
    """
    Authenticate user and return access + refresh tokens.
    """
    # Get user by email
    user = await user_repository.get_by_email(db, credentials.email)

    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password"
        )

    # Verify password
    if not verify_password(credentials.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password"
        )

    # Check if user is active
    if not user.is_active:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="User account is inactive"
        )

    # Generate tokens
    access_token = create_access_token(data={"sub": str(user.id)})
    refresh_token = create_refresh_token(data={"sub": str(user.id)})

    return TokenResponse(
        access_token=access_token,
        refresh_token=refresh_token,
        token_type="bearer"
    )

Security considerations:

✅ Generic error messages (don't reveal if email exists)
✅ Password verification before any other checks
✅ Account status validation
✅ Proper HTTP status codes

5. Token Refresh Flow

@router.post("/refresh", response_model=TokenResponse)
async def refresh_access_token(
    refresh_token: str,
    db: AsyncSession = Depends(get_db)
):
    """
    Generate new access token using refresh token.
    """
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate refresh token"
    )

    try:
        # Decode refresh token
        payload = jwt.decode(
            refresh_token,
            settings.REFRESH_SECRET_KEY,
            algorithms=[settings.ALGORITHM]
        )

        user_id: str = payload.get("sub")
        token_type: str = payload.get("type")

        if user_id is None or token_type != "refresh":
            raise credentials_exception

    except JWTError:
        raise credentials_exception

    # Verify user still exists and is active
    user = await user_repository.get_by_id(db, UUID(user_id))

    if not user or not user.is_active:
        raise credentials_exception

    # Generate new access token
    new_access_token = create_access_token(data={"sub": user_id})

    return TokenResponse(
        access_token=new_access_token,
        refresh_token=refresh_token,  # Reuse refresh token
        token_type="bearer"
    )

Why this approach?

Access tokens expire quickly (security)
Refresh tokens last longer (user experience)
User doesn't need to re-login every 15 minutes
Compromised access token has limited damage window

Real Challenge: Token Storage Strategy

The Question: Where do clients store JWT tokens?

Options I considered:

Storage Method	Pros	Cons	My Choice
localStorage	Simple, persists across tabs	XSS vulnerable	✅ Used (with CSP headers)
httpOnly Cookies	XSS safe	Needs CSRF protection, CORS complexity	Future improvement
Memory (state)	Most secure	Lost on refresh, UX issues	Not practical

For this project:

Documented the trade-offs in code comments
Implemented localStorage for simplicity
Added Content Security Policy headers
Noted httpOnly cookies as production improvement

Lesson learned: Perfect security doesn't exist. Understand trade-offs, document decisions, implement appropriate for context.

The Bug That Taught Me the Most

My first refresh token implementation worked.

It decoded the token.
It generated a new access token.
It returned 200 OK.

It was also insecure.

I forgot to check:

If the user still existed
If the user was still active

Meaning a banned user could still refresh tokens.

The fix:

user = await user_repository.get_by_id(db, UUID(user_id))

if not user or not user.is_active:
    raise HTTPException(status_code=401, detail="Invalid user")

That bug changed how I think about "working code."

Working ≠ Secure
Passing tests ≠ Safe

📦 Repository Pattern in Practice

In Spring Boot, repositories are generated for you.

In FastAPI, you write them.

Explicitly.

I implemented a generic base repository:

class BaseRepository(Generic[ModelType]):

    async def get_by_id(self, db: AsyncSession, id: UUID):
        result = await db.execute(
            select(self.model).where(self.model.id == id)
        )
        return result.scalar_one_or_none()

Then extended it:

class TaskRepository(BaseRepository[Task]):

    async def get_by_user(self, db: AsyncSession, user_id: UUID):
        result = await db.execute(
            select(Task)
            .where(Task.user_id == user_id)
            .order_by(Task.created_at.desc())
        )
        return result.scalars().all()

Benefits:

Centralized query logic
Cleaner services
Easier unit testing
Reusable across features

At first, it felt verbose.

Later, it felt disciplined.

Real Challenge: Async Session Management

The Problem I Hit:

# WRONG - This caused session closure errors
async def get_tasks(db: AsyncSession, user_id: UUID):
    tasks = await task_repository.get_by_user(db, user_id)
    # Session closes here
    for task in tasks:
        print(task.category.name)  # ERROR: Session closed!

Why? SQLAlchemy's async sessions close after the query completes. Accessing related objects (like task.category) requires the session to still be open.

The Solution:

Eager loading with selectinload:

from sqlalchemy.orm import selectinload

async def get_by_user_with_category(
    self,
    db: AsyncSession,
    user_id: UUID
) -> List[Task]:
    """Get tasks with categories eagerly loaded."""
    result = await db.execute(
        select(Task)
        .options(selectinload(Task.category))
        .where(Task.user_id == user_id)
    )
    return result.scalars().all()

Access relationships within session scope:

async def get_task_with_details(
    db: AsyncSession,
    task_id: UUID
) -> Optional[dict]:
    """Get task with all related data."""
    task = await task_repository.get_by_id(db, task_id)

    if not task:
        return None

    # Access relationships while session is active
    return {
        "id": task.id,
        "title": task.title,
        "category": task.category.name if task.category else None,
        "user": task.owner.username
    }

Lesson learned: Async requires explicit relationship loading. Plan your queries based on what data you'll need.

⚙️ Service Layer Development

Purpose of Service Layer

Why not put business logic in routes?

# BAD - Business logic in routes
@router.post("/tasks")
async def create_task(task_in: TaskCreate, db: AsyncSession = Depends(get_db)):
    # Validation logic here
    # Database operations here
    # Error handling here
    # All mixed together!

Better - Service layer:

# GOOD - Clean separation
@router.post("/tasks", response_model=TaskResponse)
async def create_task(
    task_in: TaskCreate,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db)
):
    return await task_service.create_task(db, task_in, current_user.id)

Benefits:

Routes handle HTTP concerns only
Services contain business logic
Easier to test services independently
Reusable business logic

Business logic doesn't belong in routes.

So I moved rules into services:

async def create_task(
    self,
    db: AsyncSession,
    task_in: TaskCreate,
    user_id: UUID
) -> Task:

    if task_in.due_date and task_in.due_date < datetime.utcnow():
        raise HTTPException(
            status_code=400,
            detail="Due date must be in the future"
        )

    task_data = task_in.dict()
    task_data["user_id"] = user_id

    return await task_repository.create(db, task_data)

Now:

Routes handle HTTP
Services enforce rules
Repositories manage data

The architecture from Part 1 finally paid off here.

🔧 CRUD Operations

Routes → Services → Repositories pattern in action.

What makes this production-ready?

✅ Clear documentation (auto-generated in Swagger)
✅ Input validation (Pydantic schemas)
✅ Authorization (user can only access their tasks)
✅ Proper HTTP status codes
✅ Type hints (IDE autocomplete, catch errors)
✅ Pagination (prevent large response payloads)
✅ Filtering & search (flexible queries)

Example:

@router.get("/", response_model=List[TaskResponse])
async def get_tasks(
    status: Optional[TaskStatus] = Query(None),
    search: Optional[str] = Query(None),
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db)
):
    return await task_service.get_user_tasks(
        db,
        current_user.id,
        status=status,
        search=search
    )

Underneath that simple endpoint is:

Query composition
Security enforcement
Async session handling

"Simple" rarely stays simple at scale.

🚀 Advanced Features

1. Rate Limiting with SlowAPI

Why rate limiting?

Prevent brute force attacks on auth endpoints
Protect against API abuse
Ensure fair resource usage

Configuration by endpoint type:

Auth endpoints: 5 requests/minute
Regular endpoints: 100 requests/minute
Global limit: Configurable per environment

Real challenge: SlowAPI needed custom middleware configuration on Render. Spent time debugging connection pooling and Redis integration options.

2. Advanced Filtering & Pagination

Requirements:

Filter by status, priority, category
Search in title/description
Date range queries
Pagination with metadata

Benefits:

Flexible filtering combinations
Consistent pagination structure
Metadata for building UI pagination
Type-safe responses

3. Full-Text Search

Simple but effective search implementation:

async def search_tasks(
    self,
    db: AsyncSession,
    user_id: UUID,
    search_query: str,
    skip: int = 0,
    limit: int = 100
) -> List[Task]:
    """
    Search tasks by title or description.
    Case-insensitive partial match.
    """
    search_pattern = f"%{search_query}%"

    result = await db.execute(
        select(Task)
        .where(
            and_(
                Task.user_id == user_id,
                or_(
                    Task.title.ilike(search_pattern),
                    Task.description.ilike(search_pattern)
                )
            )
        )
        .offset(skip)
        .limit(limit)
    )

    return result.scalars().all()

For future improvement:

PostgreSQL full-text search with tsvector
Search ranking/relevance scoring
Elasticsearch integration for large datasets

4. Error Handling & Logging

Centralized exception handling:

# app/core/exceptions.py
from fastapi import HTTPException

class TaskNotFoundError(HTTPException):
    def __init__(self):
        super().__init__(
            status_code=404,
            detail="Task not found"
        )

class UnauthorizedTaskAccessError(HTTPException):
    def __init__(self):
        super().__init__(
            status_code=403,
            detail="Not authorized to access this task"
        )

# app/main.py
from fastapi import Request
from fastapi.responses import JSONResponse

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    """Global exception handler for unexpected errors."""
    logger.error(f"Unhandled exception: {exc}", exc_info=True)

    return JSONResponse(
        status_code=500,
        content={
            "detail": "Internal server error",
            "path": request.url.path
        }
    )

Request/Response logging:

# app/core/monitoring.py
import time
import logging
from fastapi import Request

logger = logging.getLogger(__name__)

@app.middleware("http")
async def log_requests(request: Request, call_next):
    """Log all HTTP requests with timing."""
    start_time = time.time()

    # Log request
    logger.info(f"{request.method} {request.url.path}")

    # Process request
    response = await call_next(request)

    # Calculate duration
    duration = time.time() - start_time

    # Log response
    logger.info(
        f"{request.method} {request.url.path} "
        f"- {response.status_code} - {duration:.3f}s"
    )

    return response

Benefits:

Consistent error responses
Request tracing for debugging
Performance monitoring
Production-ready logging

🚧 Real Challenges & Solutions

Challenge 1: Async Session Closure

The Problem:

# This caused "Session is already closed" errors
async def get_task_with_category(db: AsyncSession, task_id: UUID):
    task = await task_repository.get_by_id(db, task_id)
    # Session closes after repository call
    category_name = task.category.name  # ERROR!

Why it happened:

SQLAlchemy async sessions have shorter lifecycle
Relationships not loaded by default
Accessing unloaded relationships after session closes → error

The Solution:

Eager loading:

from sqlalchemy.orm import selectinload

async def get_by_id_with_relations(
    self, 
    db: AsyncSession, 
    task_id: UUID
) -> Optional[Task]:
    result = await db.execute(
        select(Task)
        .options(
            selectinload(Task.category),
            selectinload(Task.owner)
        )
        .where(Task.id == task_id)
    )
    return result.scalar_one_or_none()

Access within session scope:

async def get_task_details(db: AsyncSession, task_id: UUID):
    task = await task_repository.get_by_id(db, task_id)

    if not task:
        return None

    # Access all relationships while session is active
    return {
        "id": task.id,
        "title": task.title,
        "category": task.category.name if task.category else None
    }

Time lost: ~3 hours debugging various "session closed" errors

Lesson learned: With async SQLAlchemy, plan your data access patterns. Decide upfront what relationships you'll need.

Challenge 2: Repository Pattern Adaptation

The Struggle:

Coming from Spring Boot's JpaRepository, I initially wrote this:

# WRONG - Trying to replicate Spring Boot directly
class TaskRepository:
    def find_by_user_id(self, user_id: UUID):  # Sync!
        # How do I make this async?

Spring Boot's repository interfaces work because of proxy magic at runtime.

FastAPI doesn't have that.

The Solution:

Accept that Python is explicit, not magic:

# RIGHT - Explicit async
class TaskRepository:
    async def get_by_user(
        self, 
        db: AsyncSession,  # Explicit session
        user_id: UUID
    ) -> List[Task]:
        result = await db.execute(
            select(Task).where(Task.user_id == user_id)
        )
        return result.scalars().all()

Lesson learned: Don't fight the language. FastAPI's explicitness is a feature, not a bug. It makes dependencies clear and testable.

Challenge 3: Token Refresh Logic

The Problem:

First implementation of token refresh had a security flaw:

What's wrong?

Doesn't check if user still exists
Doesn't check if user is still active
Could issue tokens for deleted/banned users

Time to discover: 2 days (caught during security review)

Lesson learned: JWT tokens are stateless, but security checks shouldn't be. Always validate against source of truth (database).

Challenge 4: Rate Limiting on Render

The Problem:

SlowAPI worked perfectly locally, but failed on Render:

SlowAPI: Failed to connect to Redis
Rate limiting disabled

Why?

Render's free tier doesn't include Redis.
SlowAPI uses Redis for distributed rate limiting.

The Solution:

Switch to in-memory rate limiting for free tier:

# app/core/rate_limiter.py
from slowapi import Limiter
from slowapi.util import get_remote_address
from slowapi.middleware import SlowAPIMiddleware

# Use in-memory storage (single instance)
limiter = Limiter(
    key_func=get_remote_address,
    storage_uri="memory://",  # In-memory instead of Redis
    enabled=True
)

# In main.py
app.state.limiter = limiter
app.add_middleware(SlowAPIMiddleware)

Trade-offs:

✅ Works on free tier
✅ No external dependencies
❌ Not distributed (single instance only)
❌ Resets on deployment

For production: Upgrade to Redis for distributed rate limiting across multiple instances.

Lesson learned: Free tiers have constraints. Design for your deployment environment, not just local development.

Challenge 5: Alembic Migration Conflicts

The Problem:

After developing features in parallel (tasks + categories), Alembic migrations conflicted:

alembic revision --autogenerate
Error: Multiple head revisions present

Why?

Created migrations from different branches without syncing.

The Solution:

# Merge heads
alembic merge heads -m "merge task and category features"

# Apply merged migration
alembic upgrade head

Prevention strategy:

Always pull latest migrations before creating new ones
Run alembic upgrade head before creating new migration
Use linear migration history (avoid parallel migrations)

Time lost: 1 hour debugging + fixing migration history

Lesson learned: Database migrations are sequential, not parallel. Coordinate with team (or yourself across branches).

💡 Lessons Learned

1. Async Requires Discipline

Spring Boot (blocking): Forgiving. Mix sync/async, it'll work (maybe slower).

FastAPI (async): Strict. One blocking call blocks the entire event loop.

Key rules:

Always await database calls
Use AsyncSession, not Session
Don't mix sync and async carelessly
Plan async from the start

2. Explicit is Better Than Magic

Spring Boot: Annotations do magic (@Autowired, @Transactional)

FastAPI: Dependency injection is explicit:

async def get_tasks(
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user)
):
    # Dependencies explicit in function signature

Initially: Felt verbose
After building: Appreciate the clarity

Benefit: Testability. Mock dependencies easily.

3. Repository Pattern Pays Off

Centralizing database logic in repositories:

✅ Made service layer cleaner
✅ Enabled easy testing (mock repositories)
✅ Reused queries across services
✅ Single source of truth for data access

Worth the upfront effort.

4. Security is a Process

Built authentication in iterations:

Basic password check
Added JWT tokens
Added token refresh
Added rate limiting
Added email verification
Added password reset

Each iteration revealed new attack vectors to defend against.

Security isn't a feature you add once. It's continuous hardening.

5. Documentation Writes Itself

FastAPI's auto-generated docs are incredible:

@router.post("/tasks", response_model=TaskResponse)
async def create_task(
    task_in: TaskCreate,
    current_user: User = Depends(get_current_user)
):
    """
    Create a new task.

    - **title**: Required, 1-200 characters
    - **priority**: LOW, MEDIUM, HIGH, URGENT
    """
    pass

Result: Interactive Swagger UI with:

All endpoints documented
Request/response schemas
Try-it-out functionality
Type information

No separate documentation tool needed.

🔮 What's Next: Phase 3

Coming in Part 3:

Testing

pytest setup with async support
Unit tests for services
Integration tests for API endpoints
Test coverage > 85%
Testing async code patterns

Deployment

Docker multi-stage builds
Render.com deployment
Neon PostgreSQL setup
Environment variable management
Health checks & monitoring

Production Features

Performance metrics
Uptime tracking
Cost analysis ($0 free tier!)
CI/CD pipeline
Production best practices

Follow me to get notified when Part 3 drops!

📝 Complete Code & Resources

GitHub Repository:

🔗 https://github.com/ravigupta97/task_management_api
Fully documented code
Clean commit history
README with setup instructions

Live API Documentation:

🚀 https://task-management-api-a775.onrender.com/docs
Interactive Swagger UI
Try all endpoints
See request/response schemas

Key Files to Explore:

app/core/security.py - JWT implementation
app/repositories/ - Repository pattern
app/services/ - Business logic layer
app/api/v1/ - Route handlers

🙏 Acknowledgments

Tools & Resources:

FastAPI documentation (exceptional quality)
SQLAlchemy 2.0 docs (comprehensive)
AI assistants (concept clarification)
Open-source community

What helped most:

Reading production FastAPI codebases on GitHub
AI tools for understanding complex concepts
Trial and error (every bug taught something)

💬 Discussion

Questions for you:

How do you handle async session management?
What's your approach to API authentication?
Repository pattern: overkill or essential?
Biggest challenge you've faced with FastAPI?

Let's learn together - comment below! 👇

This is Part 2 of 3 in my FastAPI journey.

Part 1: Architecture & Design
Part 3: Testing & Deployment (coming soon)

Connect with me:

LinkedIn: [https://www.linkedin.com/in/ravigupta97/]
GitHub: [https://github.com/ravigupta97]

#FastAPI #Python #Backend #JWT #Authentication #API #CleanArchitecture #WebDevelopment #SoftwareEngineering