DEV Community: archi-jain

Day 8/100: API Documentation & Error Handling - The Professional Polish

archi-jain — Thu, 12 Mar 2026 15:52:16 +0000

Part of my 100 Days of Code journey. Today we transform working code into production-ready API.

The Challenge

Add comprehensive documentation and standardized error handling to make the API developer-friendly and production-ready.

The Problem:
My API works great... for me. But what about:

New developers trying to integrate?
Frontend developers needing to consume the API?
DevOps needing to monitor health?
Users getting cryptic error messages?

The Solution:
Professional documentation, structured errors, CORS support, and health monitoring.

Why Documentation Matters

True story: I once worked with an API that had zero documentation. Every endpoint was a mystery:

What parameters does it accept?
What response format does it return?
What errors can it throw?
How do I authenticate?

Result:

3 weeks to integrate (should've been 3 days)
Countless support emails
Trial-and-error development
Frustrated developers

Good documentation changes everything.

Step-by-Step Implementation

Step 1: Enhanced FastAPI Configuration

from fastapi import FastAPI

app = FastAPI(
    title="Task Management API",
    description="""
    ## 🚀 Complete Task Management System

    A production-ready REST API for managing tasks with authentication, 
    tags, priorities, due dates, and advanced filtering.

    ### Features

    * **🔐 JWT Authentication** - Secure user registration and login
    * **📋 Task Management** - Full CRUD operations with rich metadata
    * **🏷️ Tag System** - Organize tasks with custom tags and colors
    * **⚡ Priority Levels** - High, medium, low priorities
    * **📅 Due Dates** - Track deadlines and overdue tasks
    * **🔍 Advanced Filtering** - Search, sort, filter by multiple criteria
    * **📊 Analytics** - Task statistics and insights
    * **⚙️ Bulk Operations** - Efficient multi-task updates

    ### Getting Started

    1. Register a new account at `/auth/register`
    2. Login at `/auth/login` to get your access token
    3. Click the 🔒 Authorize button and paste your token
    4. Start managing your tasks!
    """,
    version="1.0.0",
    contact={
        "name": "Archi Jain",
        "url": "https://github.com/archi-jain/100-days-of-python",
        "email": "your-email@example.com"
    },
    license_info={
        "name": "MIT License",
        "url": "https://opensource.org/licenses/MIT"
    }
)

What this does:

When developers visit /docs, they see:

Clear API overview
Feature list
Getting started guide
Contact information
License details

Professional first impression!

Step 2: Endpoint Organization with Tags

app = FastAPI(
    # ... previous config ...
    openapi_tags=[
        {
            "name": "Authentication",
            "description": "User registration, login, and profile management"
        },
        {
            "name": "Tasks",
            "description": "Task CRUD operations, filtering, sorting, and bulk actions"
        },
        {
            "name": "Tags",
            "description": "Tag management for organizing tasks"
        },
        {
            "name": "System",
            "description": "API health checks and system information"
        }
    ]
)

Result:
Endpoints grouped logically in documentation. Developers can find what they need quickly.

Step 3: Detailed Endpoint Documentation

Before:

@app.post("/tasks")
def create_task(task: TaskCreate, ...):
    # code

After:

@router.post(
    "/",
    response_model=schemas.TaskResponse,
    status_code=status.HTTP_201_CREATED,
    summary="Create a new task",
    description="""
    Create a new task with optional tags, priority, and due date.

    **Request Body:**
    - **title** (required): Task name/description
    - **description** (optional): Detailed information
    - **priority** (optional): high, medium, or low (default: medium)
    - **due_date** (optional): Deadline in ISO 8601 format
    - **tag_ids** (optional): List of tag UUIDs to associate

    **Returns:**
    - Newly created task with generated ID and timestamps

    **Errors:**
    - 400: Invalid tag IDs or past due date
    - 401: Not authenticated
    - 422: Validation error (missing title, invalid format)
    """,
    responses={
        201: {
            "description": "Task created successfully",
            "content": {
                "application/json": {
                    "example": {
                        "id": "123e4567-e89b-12d3-a456-426614174000",
                        "title": "Complete project",
                        "completed": False,
                        "priority": "high",
                        "due_date": "2026-03-15T17:00:00Z"
                    }
                }
            }
        }
    }
)
def create_task(task: schemas.TaskCreate, ...):
    """Create a new task for the authenticated user"""
    # code

What developers see:

Clear summary
Detailed description
Parameter explanations
Success example
Error cases
Sample request/response

Step 4: Schema Examples

class TaskCreate(BaseModel):
    title: str = Field(
        ..., 
        max_length=100,
        description="Task title (required)",
        examples=["Complete project proposal", "Review pull requests"]
    )
    description: Optional[str] = Field(
        None, 
        max_length=500,
        description="Detailed task description (optional)",
        examples=["Prepare Q1 financial review slides"]
    )

    model_config = {
        "json_schema_extra": {
            "examples": [
                {
                    "title": "Submit quarterly report",
                    "description": "Prepare Q1 2026 financial report",
                    "priority": "high",
                    "due_date": "2026-03-15T17:00:00Z"
                }
            ]
        }
    }

Result:
When developers click "Try it out", the form is pre-filled with valid example data!

Step 5: CORS Configuration

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "http://localhost:3000",  # React default
        "http://localhost:5173",  # Vite default
        "http://localhost:8080",  # Vue default
        # Add your production domains
    ],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Why this matters:

Without CORS:

// Frontend code
fetch('http://localhost:8000/tasks')

// Browser blocks:
// "CORS policy: No 'Access-Control-Allow-Origin' header"

With CORS:

fetch('http://localhost:8000/tasks')
// ✅ Works!

Frontend developers can integrate immediately.

Step 6: Standardized Error Responses

Before:

if not task:
    raise HTTPException(404, "Task not found")

# Response: {"detail": "Task not found"}

After:

class APIError(HTTPException):
    def __init__(self, status_code, error_code, message, details=None):
        super().__init__(
            status_code=status_code,
            detail={
                "error": {
                    "code": error_code,
                    "message": message,
                    "details": details,
                    "timestamp": datetime.utcnow().isoformat()
                }
            }
        )

class TaskNotFoundError(APIError):
    def __init__(self, task_id: str):
        super().__init__(
            status_code=404,
            error_code="TASK_NOT_FOUND",
            message="Task not found",
            details=f"Task with ID {task_id} does not exist or you don't have access"
        )

# Usage:
if not task:
    raise TaskNotFoundError(str(task_id))

# Response:
{
  "error": {
    "code": "TASK_NOT_FOUND",
    "message": "Task not found",
    "details": "Task with ID abc123 does not exist or you don't have access",
    "timestamp": "2026-03-08T10:30:00Z"
  }
}

Benefits:

Error codes - Frontend can handle programmatically
Detailed messages - Developers know what went wrong
Timestamps - Debugging and logging
Consistency - All errors follow same format

Frontend handling:

try {
  const response = await fetch('/tasks/invalid-id')
  const data = await response.json()

  if (!response.ok) {
    switch (data.error.code) {
      case 'TASK_NOT_FOUND':
        showNotification("Task doesn't exist")
        break
      case 'AUTHENTICATION_FAILED':
        redirectToLogin()
        break
      default:
        showError(data.error.message)
    }
  }
} catch (error) {
  // Handle error
}

Step 7: Custom Validation Error Handler

from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    errors = []
    for error in exc.errors():
        errors.append({
            "field": " -> ".join(str(x) for x in error["loc"]),
            "message": error["msg"],
            "type": error["type"]
        })

    return JSONResponse(
        status_code=422,
        content={
            "error": {
                "code": "VALIDATION_ERROR",
                "message": "Request validation failed",
                "details": errors,
                "timestamp": datetime.utcnow().isoformat()
            }
        }
    )

Before:

{
  "detail": [
    {
      "loc": ["body", "title"],
      "msg": "field required",
      "type": "value_error.missing"
    }
  ]
}

After:

{
  "error": {
    "code": "VALIDATION_ERROR",
    "message": "Request validation failed",
    "details": [
      {
        "field": "body -> title",
        "message": "field required",
        "type": "value_error.missing"
      }
    ],
    "timestamp": "2026-03-08T10:30:00Z"
  }
}

Much clearer!

Step 8: Health Check Endpoint

@app.get("/health", tags=["System"])
def health_check():
    """Check API health and database connectivity"""
    from database import SessionLocal

    db_status = "healthy"
    try:
        db = SessionLocal()
        db.execute("SELECT 1")
        db.close()
    except Exception as e:
        db_status = f"unhealthy: {str(e)}"

    return {
        "status": "healthy" if db_status == "healthy" else "degraded",
        "version": "1.0.0",
        "timestamp": datetime.utcnow().isoformat(),
        "database": db_status
    }

Why DevOps loves this:

# Kubernetes health check
livenessProbe:
  httpGet:
    path: /health
    port: 8000
  initialDelaySeconds: 30
  periodSeconds: 10

# Monitoring alert
if health_status != "healthy":
  send_alert()

Step 9: API Information Endpoint

@app.get("/", tags=["System"])
def root():
    """Get basic API information"""
    return {
        "name": "Task Management API",
        "version": "1.0.0",
        "status": "operational",
        "documentation": {
            "swagger": "/docs",
            "redoc": "/redoc"
        },
        "endpoints": {
            "health": "/health",
            "auth": "/auth",
            "tasks": "/tasks",
            "tags": "/tags"
        },
        "features": [
            "JWT Authentication",
            "Task Management",
            "Tags & Priorities",
            "Due Dates",
            "Advanced Filtering",
            "Bulk Operations"
        ]
    }

First thing developers see:

curl http://localhost:8000/

# Returns clear API overview with links to docs

Testing the Enhancements

Test 1: Enhanced Documentation

# Open browser
http://localhost:8000/docs

What you'll see:

Professional API overview
Organized endpoint groups
Detailed descriptions
Try-it-out with examples
Error documentation

Test 2: Health Check

curl http://localhost:8000/health

Response:

{
  "status": "healthy",
  "version": "1.0.0",
  "timestamp": "2026-03-08T10:30:00.123456",
  "database": "healthy"
}

Test 3: Structured Error

# Create task with missing title
curl -X POST http://localhost:8000/tasks \
  -H "Authorization: Bearer TOKEN" \
  -H "Content-Type: application/json" \
  -d '{}'

Response:

{
  "error": {
    "code": "VALIDATION_ERROR",
    "message": "Request validation failed",
    "details": [
      {
        "field": "body -> title",
        "message": "field required",
        "type": "value_error.missing"
      }
    ],
    "timestamp": "2026-03-08T10:30:00Z"
  }
}

Clear, actionable error!

Test 4: CORS from Frontend

// React component
useEffect(() => {
  fetch('http://localhost:8000/tasks', {
    headers: {
      'Authorization': `Bearer ${token}`
    }
  })
  .then(res => res.json())
  .then(data => setTasks(data))
  // ✅ Works! No CORS errors
}, [])

Real-World Impact

Before Day 8:

New Developer Experience:

Opens /docs - sees basic endpoint list
Tries to create task - no examples
Gets error: {"detail": "Error"} - what error?
Frontend can't connect - CORS errors
No way to check if API is running
Integration takes 2 weeks

Support burden:

"How do I create a task?"
"What's the error response format?"
"Why am I getting CORS errors?"
"Is the API down?"

After Day 8:

New Developer Experience:

Opens /docs - sees professional overview
Reads detailed endpoint descriptions
Clicks "Try it out" - sees example pre-filled
Gets structured error with clear message
Frontend connects immediately (CORS configured)
/health endpoint confirms API status
Integration takes 2 days

Support burden:

Developers self-serve via documentation
Clear errors reduce confusion
CORS works out of the box
Health check for monitoring

Key Takeaways

1. Documentation is a Feature

Code that works but isn't documented is only half-done. Documentation is how your API gets adopted.

2. Errors Should Help, Not Confuse

# Bad
{"detail": "Error"}

# Good
{
  "error": {
    "code": "TASK_NOT_FOUND",
    "message": "Task not found",
    "details": "Task abc123 does not exist or you lack access",
    "timestamp": "2026-03-08T10:30:00Z"
  }
}

3. CORS is Not Optional

If you want frontend developers to use your API, CORS must be configured. Period.

4. Health Checks Enable Monitoring

DevOps can't monitor what they can't query. /health endpoint = observable systems.

5. Examples Save Time

One good example is worth a thousand words. Pre-filled "Try it out" = instant understanding.

What I Learned

Technical Skills

✅ OpenAPI/Swagger configuration

✅ FastAPI metadata and descriptions

✅ Custom exception handlers

✅ CORS middleware setup

✅ Health check implementation

✅ Schema examples and validation

✅ Error response standardization

API Design Principles

✅ Developer experience is crucial

✅ Consistency reduces friction

✅ Good errors save support time

✅ Documentation = adoption

Production Readiness

✅ CORS for frontend integration

✅ Health checks for monitoring

✅ Structured errors for clients

✅ Comprehensive documentation

The Numbers

Before:

Basic documentation: 5/10
Error clarity: 3/10
Frontend ready: 0/10 (CORS blocked)
Monitoring: 0/10 (no health check)

After:

Professional documentation: 10/10
Error clarity: 10/10
Frontend ready: 10/10 (CORS configured)
Monitoring: 10/10 (health endpoint)

Time investment: 2.5 hours

Value added: Immeasurable

Tomorrow's Plans

Day 9 will add:

CSV/JSON export functionality
Data import from files
Batch task creation
Backup/restore capabilities

But today? Today I celebrate making my API production-ready and developer-friendly! 🎉

Resources

Full Code

Complete code on GitHub:
100-days-of-python/days/008

Time Investment: 2.5 hours

Knowledge Gained: Professional API development practices

Feeling: Like a real API developer 🚀

Day 8/100 complete! ✅

Tomorrow: Data import/export features

Following my journey?

📝 Blog | 🐦 Twitter | 💼 LinkedIn

Tags: #100DaysOfCode #Python #FastAPI #APIDevelopment #Documentation #OpenAPI #Swagger #ErrorHandling #CORS #DeveloperExperience #ProductionReady #LearningInPublic

Day 6/100: Quick Win - Task Priorities in 1 Hour

archi-jain — Mon, 09 Mar 2026 18:28:09 +0000

Sometimes the best features are the simplest.

The Challenge

Add task priority levels (high/medium/low) with filtering, sorting, and statistics.

Time constraint: Only 1 hour available today.

Result: Fully implemented in 70 minutes!

Why Priorities Matter

User problem: "All my tasks look the same. Which should I do first?"

Solution: Let users mark priority levels.

Impact:

Focus on high priority first
Filter out low priority noise
See if too many tasks marked urgent (priority inflation)

Implementation (15 Lines of Code)

1. Add Enum

class PriorityEnum(str, Enum):
    low = "low"
    medium = "medium"
    high = "high"

2. Add Column

priority = Column(String(10), default="medium", index=True)

3. Update Schemas

priority: Optional[PriorityEnum] = Field(default=PriorityEnum.medium)

4. Add Filter

if priority:
    query = query.filter(Task.priority == priority)

5. Update Stats

"by_priority": {
    "high": tasks.filter(Task.priority == "high").count(),
    "medium": tasks.filter(Task.priority == "medium").count(),
    "low": tasks.filter(Task.priority == "low").count()
}

Done!

Query examples:

GET /tasks?priority=high
GET /tasks?priority=high&completed=false
GET /tasks?sort_by=priority&order=desc

Key Takeaways

✅ Enums prevent typos (no "hgh" or "medum")

✅ Defaults prevent decision fatigue

✅ Indexes make filtering fast

✅ Simple features can have big impact

Time: 1 hour

Lines changed: ~15

User value: Immediate

Sometimes less IS more!

Day 6/100 complete! ✅

Following my journey?

📝 Blog | 🐦 Twitter | 💼 LinkedIn

100DaysOfCode #Python #FastAPI #QuickWins

Day 5/100: Many-to-Many Relationships - Tags, Associations, and Bulk Operations

archi-jain — Sun, 08 Mar 2026 17:12:56 +0000

Part of my 100 Days of Code journey. Today we tackle one of the most important database patterns.

The Challenge

Implement a tagging system with many-to-many relationships and add bulk operations for efficiency.

The Problem: Tasks need flexible organization. Fixed categories don't work because:

One task can belong to multiple categories (urgent AND work AND client-facing)
Categories can have multiple tasks
Users want custom labels, not predefined options

The Solution: Many-to-many relationships via junction tables, plus bulk operations for power users.

Why Many-to-Many Matters

One-to-Many relationships (what we've used so far):

User ──┬─→ Task 1
       ├─→ Task 2
       └─→ Task 3

Each task has ONE user. Simple!

Many-to-Many relationships (what we need):

Task 1 ──┬─→ Tag: urgent
         └─→ Tag: work

Task 2 ──┬─→ Tag: urgent
         ├─→ Tag: personal
         └─→ Tag: home

Tag: urgent ──┬─→ Task 1
              └─→ Task 2

Problem: Can't use a simple foreign key!

If we add tag_id to tasks table:

❌ Task can only have ONE tag
❌ Need multiple tag_id columns (tag_id_1, tag_id_2...) - terrible design!

Solution: Association table (junction table)

Understanding Association Tables

An association table (also called junction table or link table) connects two tables in a many-to-many relationship.

Three tables:

tasks table:
| id | title | owner_id |
|----|-------|----------|
| 1 | Meeting | user-a |
| 2 | Report | user-a |
| 3 | Email | user-a |

tags table:
| id | name | color | owner_id |
|----|------|-------|----------|
| A | urgent | #FF0000 | user-a |
| B | work | #0000FF | user-a |
| C | personal | #00FF00 | user-a |

task_tags table (association):
| task_id | tag_id |
|---------|--------|
| 1 | A |
| 1 | B |
| 2 | A |
| 3 | C |

What this means:

Task 1 (Meeting) has tags: urgent, work
Task 2 (Report) has tags: urgent
Task 3 (Email) has tags: personal
Tag A (urgent) is on: Task 1, Task 2
Tag B (work) is on: Task 1

Perfect flexibility!

Step-by-Step Implementation

Step 1: Creating the Association Table

from sqlalchemy import Table, Column, ForeignKey
from sqlalchemy.dialects.postgresql import UUID

task_tags = Table(
    "task_tags",
    Base.metadata,
    Column("task_id", UUID(as_uuid=True), ForeignKey("tasks.id", ondelete="CASCADE"), primary_key=True),
    Column("tag_id", UUID(as_uuid=True), ForeignKey("tags.id", ondelete="CASCADE"), primary_key=True),
)

Breaking it down:

Why Table() instead of a class?

# Simple association (just links)
task_tags = Table(...)  # Recommended!

# Complex association (with extra fields like created_at, priority)
class TaskTag(Base):  # Use when needed
    __tablename__ = "task_tags"
    task_id = Column(...)
    tag_id = Column(...)
    priority = Column(Integer)
    created_at = Column(DateTime)

For pure many-to-many (just linking), Table() is cleaner.

Composite primary key:

primary_key=True  # On task_id
primary_key=True  # On tag_id

Both together form the primary key! This means:

✅ Can't add same tag twice to same task
✅ (task_id=1, tag_id=A) can exist once
✅ (task_id=1, tag_id=B) is different, allowed

CASCADE deletes:

ForeignKey("tasks.id", ondelete="CASCADE")

What happens:

Task deleted → All task_tags entries with that task_id deleted
Tag deleted → All task_tags entries with that tag_id deleted
No orphaned associations!

Without CASCADE:

Task 1 deleted
task_tags still has: (task_id=1, tag_id=A)  # Orphan!
Next query: "Show tasks with tag A"
Result: Error! task_id=1 doesn't exist!

With CASCADE:

Task 1 deleted
Database automatically deletes: (task_id=1, tag_id=A), (task_id=1, tag_id=B)
Clean! No orphans!

Step 2: Creating the Tag Model

class Tag(Base):
    __tablename__ = "tags"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    name = Column(String(50), nullable=False)
    color = Column(String(7))  # Hex color: #RRGGBB
    created_at = Column(DateTime, default=datetime.utcnow)
    owner_id = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"))

    # Relationships
    owner = relationship("User", back_populates="tags")
    tasks = relationship("Task", secondary=task_tags, back_populates="tags")

    # Unique constraint: user can't have duplicate tag names
    __table_args__ = (
        Index("idx_owner_tag", "owner_id", "name", unique=True),
    )

Understanding the relationship:

tasks = relationship("Task", secondary=task_tags, back_populates="tags")

"Task": Links to Task model
secondary=task_tags: Use task_tags as junction table
back_populates="tags": Links to tags field in Task model

The magic:

tag = db.query(Tag).first()
print(tag.tasks)  # [Task(...), Task(...)]

# SQLAlchemy auto-generates SQL:
# SELECT tasks.* 
# FROM tasks 
# JOIN task_tags ON tasks.id = task_tags.task_id
# WHERE task_tags.tag_id = 'tag-uuid'

We didn't write that JOIN! SQLAlchemy did!

Composite unique constraint:

Index("idx_owner_tag", "owner_id", "name", unique=True)

What it prevents:

User A creates tag "urgent" ✅
User A creates tag "urgent" again ❌ Error: duplicate
User B creates tag "urgent" ✅ Different user, allowed!

Step 3: Updating the Task Model

class Task(Base):
    __tablename__ = "tasks"

    # ... existing fields ...

    # Relationships
    owner = relationship("User", back_populates="tasks")
    tags = relationship("Tag", secondary=task_tags, back_populates="tasks")

Bidirectional relationship:

From Task:

task = db.query(Task).first()
print(task.tags)  # [Tag(name="urgent"), Tag(name="work")]

From Tag:

tag = db.query(Tag).filter(Tag.name == "urgent").first()
print(tag.tasks)  # [Task(title="Meeting"), Task(title="Report")]

Both directions work!

Step 4: Creating Tags

@router.post("/tags", response_model=TagResponse)
def create_tag(tag: TagCreate, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
    # Check if tag already exists for this user
    existing = db.query(Tag).filter(
        Tag.owner_id == current_user.id,
        Tag.name == tag.name
    ).first()

    if existing:
        raise HTTPException(400, "Tag already exists")

    new_tag = Tag(
        name=tag.name,
        color=tag.color,
        owner_id=current_user.id
    )

    db.add(new_tag)
    db.commit()
    db.refresh(new_tag)

    return new_tag

Why check for existing tag?

Without check:

POST /tags {"name": "urgent", "color": "#FF0000"}  ✅
POST /tags {"name": "urgent", "color": "#0000FF"}  💥 Database error!

Database throws error because of unique constraint, but:

Error message is cryptic
HTTP 500 instead of 400
Bad user experience

With check:

POST /tags {"name": "urgent"}  ✅
POST /tags {"name": "urgent"}  ✅ 400 Bad Request "Tag already exists"

Clean error handling!

Step 5: Creating Tasks with Tags

@router.post("/tasks", response_model=TaskResponse)
def create_task(task: TaskCreate, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
    new_task = Task(
        title=task.title,
        description=task.description,
        owner_id=current_user.id
    )

    # Add tags if provided
    if task.tag_ids:
        tags = db.query(Tag).filter(
            Tag.id.in_(task.tag_ids),
            Tag.owner_id == current_user.id
        ).all()

        if len(tags) != len(task.tag_ids):
            raise HTTPException(400, "One or more tags not found")

        new_task.tags = tags

    db.add(new_task)
    db.commit()
    db.refresh(new_task)

    return new_task

Understanding id.in_():

Tag.id.in_(["uuid1", "uuid2", "uuid3"])

SQL generated:

SELECT * FROM tags 
WHERE id IN ('uuid1', 'uuid2', 'uuid3')
  AND owner_id = 'current-user-uuid'

One query gets all tags!

Setting tags:

new_task.tags = tags

What SQLAlchemy does:

Creates task in tasks table
For each tag, creates entry in task_tags table:
- INSERT INTO task_tags (task_id, tag_id) VALUES (new_task.id, tag.id)

All automatic!

Validation:

if len(tags) != len(task.tag_ids):
    raise HTTPException(400, "One or more tags not found")

If user sends 3 tag IDs but only 2 exist → error!

Step 6: Filtering Tasks by Tags

@router.get("/tasks")
def get_tasks(
    tag_id: Optional[UUID] = Query(None),
    tag_name: Optional[str] = Query(None),
    ...
):
    query = db.query(Task).filter(Task.owner_id == current_user.id)

    # Filter by tag ID
    if tag_id:
        query = query.filter(Task.tags.any(Tag.id == tag_id))

    # Filter by tag name
    if tag_name:
        query = query.filter(Task.tags.any(Tag.name == tag_name.lower()))

    return query.all()

Understanding .any():

Task.tags.any(Tag.id == tag_id)

Translation: "Get tasks where ANY of their tags has this ID"

SQL generated:

SELECT tasks.* 
FROM tasks
WHERE tasks.owner_id = 'user-uuid'
  AND EXISTS (
    SELECT 1 
    FROM task_tags 
    JOIN tags ON tags.id = task_tags.tag_id
    WHERE task_tags.task_id = tasks.id
      AND tags.id = 'tag-uuid'
  )

Example:

# Get all tasks tagged "urgent"
GET /tasks?tag_name=urgent

# Get all urgent tasks that are incomplete
GET /tasks?tag_name=urgent&completed=false

# Get all work tasks, sorted by date
GET /tasks?tag_name=work&sort_by=created_at&order=desc

Filters stack beautifully!

Step 7: Adding/Removing Tags from Tasks

Add tags:

@router.post("/tasks/{task_id}/tags")
def add_tags_to_task(task_id: UUID, tag_data: BulkUpdateTags, ...):
    task = db.query(Task).filter(
        Task.id == task_id,
        Task.owner_id == current_user.id
    ).first()

    if not task:
        raise HTTPException(404, "Task not found")

    tags = db.query(Tag).filter(
        Tag.id.in_(tag_data.tag_ids),
        Tag.owner_id == current_user.id
    ).all()

    for tag in tags:
        if tag not in task.tags:
            task.tags.append(tag)

    db.commit()
    db.refresh(task)

    return task

Why check if tag not in task.tags?

Without check:

task.tags.append(tag)  # Adds tag
task.tags.append(tag)  # Adds again!
# Database error: duplicate primary key (task_id, tag_id)

With check:

if tag not in task.tags:
    task.tags.append(tag)  # Safe!

Remove tag:

@router.delete("/tasks/{task_id}/tags/{tag_id}")
def remove_tag_from_task(task_id: UUID, tag_id: UUID, ...):
    task = ...
    tag = ...

    if tag in task.tags:
        task.tags.remove(tag)
        db.commit()

    return task

What remove() does:

task.tags.remove(tag)

SQLAlchemy:

DELETE FROM task_tags 
WHERE task_id = 'task-uuid' 
  AND tag_id = 'tag-uuid'

Association removed!

Step 8: Bulk Operations

Bulk complete:

@router.post("/tasks/bulk/complete")
def bulk_complete_tasks(bulk_data: BulkTaskIds, ...):
    updated = db.query(Task).filter(
        Task.id.in_(bulk_data.task_ids),
        Task.owner_id == current_user.id
    ).update({"completed": True}, synchronize_session=False)

    db.commit()

    return {
        "updated": updated,
        "message": f"{updated} tasks marked complete"
    }

Why bulk operations matter:

Without bulk (N requests):

// Frontend makes 50 API calls
taskIds.forEach(id => {
  fetch(`/tasks/${id}`, {
    method: 'PATCH',
    body: JSON.stringify({completed: true})
  })
})

// Result:
// - 50 HTTP requests
// - 50 SQL queries
// - ~5 seconds total

With bulk (1 request):

// Frontend makes 1 API call
fetch('/tasks/bulk/complete', {
  method: 'POST',
  body: JSON.stringify({task_ids: taskIds})
})

// Result:
// - 1 HTTP request
// - 1 SQL query
// - ~0.1 seconds total

50x faster!

The SQL:

UPDATE tasks 
SET completed = true 
WHERE id IN ('uuid1', 'uuid2', ..., 'uuid50')
  AND owner_id = 'user-uuid'

One query updates 50 tasks!

synchronize_session=False:

.update({...}, synchronize_session=False)

What it means:

We're doing bulk update
Don't update ORM objects in memory
Faster (we don't need those objects anyway)

Use when you don't need the updated objects returned.

Bulk delete:

@router.delete("/tasks/bulk")
def bulk_delete_tasks(bulk_data: BulkTaskIds, ...):
    deleted = db.query(Task).filter(
        Task.id.in_(bulk_data.task_ids),
        Task.owner_id == current_user.id
    ).delete(synchronize_session=False)

    db.commit()

    return {
        "deleted": deleted,
        "message": f"{deleted} tasks deleted"
    }

SQL:

DELETE FROM tasks 
WHERE id IN ('uuid1', 'uuid2', 'uuid3')
  AND owner_id = 'user-uuid'

CASCADE in action:

DELETE FROM tasks WHERE id = 'uuid1'
  ↓ CASCADE
DELETE FROM task_tags WHERE task_id = 'uuid1'

All associations cleaned up automatically!

Testing the Features

Test 1: Create Tags

curl -X POST "http://localhost:8000/tags" \
  -H "Authorization: Bearer TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"name": "urgent", "color": "#FF0000"}'

curl -X POST "http://localhost:8000/tags" \
  -H "Authorization: Bearer TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"name": "work", "color": "#0000FF"}'

Test 2: Create Task with Tags

curl -X POST "http://localhost:8000/tasks" \
  -H "Authorization: Bearer TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "title": "Team Meeting",
    "tag_ids": ["TAG_UUID_1", "TAG_UUID_2"]
  }'

Test 3: Filter by Tags

# All urgent tasks
curl -H "Authorization: Bearer TOKEN" \
  "http://localhost:8000/tasks?tag_name=urgent"

# Urgent AND incomplete
curl -H "Authorization: Bearer TOKEN" \
  "http://localhost:8000/tasks?tag_name=urgent&completed=false"

Test 4: Bulk Complete

curl -X POST "http://localhost:8000/tasks/bulk/complete" \
  -H "Authorization: Bearer TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"task_ids": ["UUID1", "UUID2", "UUID3"]}'

Response:

{
  "updated": 3,
  "message": "3 tasks marked complete"
}

Real-World Impact

Before tagging:

Tasks in flat list
No organization
Hard to find related tasks

After tagging:

Flexible organization
Filter by project, priority, context
Visual coding with colors
Power user features (bulk ops)

Example workflows:

# Show all urgent client tasks
GET /tasks?tag_name=urgent&tag_name=client

# Complete all daily standup tasks
POST /tasks/bulk/complete
Body: {task_ids: [all standup task IDs]}

# Find all bugs
GET /tasks?tag_name=bug&completed=false

Performance Considerations

Junction table indexes:

# Automatically indexed (part of primary key)
- (task_id, tag_id)

# Should also index individually for reverse lookups
Index("idx_task_tags_task", "task_id")
Index("idx_task_tags_tag", "tag_id")

Query performance:

Without indexes:

SELECT tasks.* FROM tasks
JOIN task_tags ON ...
WHERE tag_id = 'uuid'
-- Scans entire task_tags table (~10,000 rows: 500ms)

With indexes:

-- Same query, uses index
-- (~10,000 rows: 50ms)

10x faster with proper indexes!

What I Learned

Technical Skills

✅ Many-to-many relationship implementation

✅ Association tables (junction tables)

✅ Composite primary keys

✅ SQLAlchemy secondary parameter

✅ Relationship navigation (bidirectional)

✅ .any() for filtering relationships

✅ Bulk update/delete operations

✅ CASCADE behavior and cleanup

Database Design

✅ When to use junction tables

✅ Composite unique constraints

✅ Index optimization for associations

✅ Preventing orphaned records

API Design

✅ Bulk operation endpoints

✅ Tag-based filtering

✅ Flexible organization systems

✅ Power user features

Common Mistakes I Avoided

Mistake 1: Forgetting CASCADE

# Bad - orphaned associations
ForeignKey("tasks.id")

# Good - clean cascade
ForeignKey("tasks.id", ondelete="CASCADE")

Mistake 2: Not Validating Ownership

# Bad - user can add anyone's tags
tags = db.query(Tag).filter(Tag.id.in_(tag_ids)).all()

# Good - only user's tags
tags = db.query(Tag).filter(
    Tag.id.in_(tag_ids),
    Tag.owner_id == current_user.id
).all()

Mistake 3: Duplicate Tag Names

# Bad - no uniqueness check
Tag(name="urgent")  # User creates
Tag(name="urgent")  # User creates again - confusing!

# Good - unique constraint
Index("idx_owner_tag", "owner_id", "name", unique=True)

Mistake 4: N Loop Updates

# Bad - 50 queries
for task_id in task_ids:
    task = db.query(Task).get(task_id)
    task.completed = True
    db.commit()

# Good - 1 query
db.query(Task).filter(Task.id.in_(task_ids)).update({...})

Tomorrow's Plans

Day 6 will add:

Due dates and reminders
Task priorities (high, medium, low)
Recurring tasks
Calendar view support

But today? Today I celebrate mastering many-to-many relationships - one of the most important database patterns! 🎉

Resources

Full Code

Complete code on GitHub:
100-days-of-python/days/005

Time Investment: 3 hours

Knowledge Gained: Advanced relationship patterns

Feeling: Like I understand how real databases work 🗄️

Day 5/100 complete! ✅

Tomorrow: Due dates, priorities, and scheduling

Following my journey?

📝 Blog | 🐦 Twitter | 💼 LinkedIn

Tags: #100DaysOfCode #Python #FastAPI #PostgreSQL #SQLAlchemy #ManyToMany #JunctionTables #BulkOperations #DatabaseDesign #LearningInPublic

Day 4/100: Advanced Query Features - From Basic CRUD to Production API

archi-jain — Sun, 08 Mar 2026 03:58:45 +0000

Part of my 100 Days of Code journey. Today we make the API actually usable at scale.

The Challenge

Add advanced query features to the Task Management API: filtering, sorting, pagination, search, and statistics.

The Problem: Yesterday's API works, but imagine having 1,000 tasks:

Can't find specific tasks (no search)
Loading all 1,000 at once (slow, wasteful)
Can't sort by priority (no sorting)
No insights into completion rates (no analytics)

The Solution: Implement production-grade query features with proper indexing.

Why Advanced Queries Matter

Real scenario: You have 5,000 tasks in your database.

Without these features:

# User clicks "View Tasks"
tasks = db.query(Task).all()  # Loads ALL 5,000 tasks!
# Result: 
# - 3 second page load
# - 5MB data transfer
# - Angry users

With these features:

# User searches "meeting", page 1
tasks = db.query(Task)\
    .filter(Task.title.ilike("%meeting%"))\
    .order_by(desc(Task.created_at))\
    .limit(20)
# Result:
# - 0.05 second response
# - 50KB data transfer
# - Happy users

This is the difference between a toy and a product.

What We're Building

Query Features:

Filtering - Get only completed or pending tasks
Sorting - Order by date, title, or status
Pagination - Load 20 tasks at a time
Search - Find tasks by keyword
Combined - Use all features together

Statistics:

Total task count
Completion rate
Tasks created today/this week

Performance:

Database indexes for speed
Query constraints for safety

Understanding Query Parameters

Query parameters are the ?key=value parts of URLs:

https://api.example.com/tasks?completed=false&sort_by=created_at&limit=20
                              ↑              ↑                 ↑
                           Filter          Sort            Paginate

In FastAPI:

from fastapi import Query

@app.get("/tasks")
def get_tasks(
    completed: Optional[bool] = Query(None),
    sort_by: str = Query("created_at"),
    skip: int = Query(0, ge=0),
    limit: int = Query(20, ge=1, le=100)
):
    # FastAPI auto-extracts and validates!

What FastAPI does automatically:

Extracts completed from URL
Converts "true" string → Python True boolean
Validates skip >= 0 and limit between 1-100
Returns 422 error if validation fails

Beautiful!

Step-by-Step Implementation

Step 1: Query Parameter Setup

from fastapi import Query
from typing import Optional

@router.get("/", response_model=List[schemas.TaskResponse])
def get_tasks(
    # Filtering
    completed: Optional[bool] = Query(None, description="Filter by status"),

    # Sorting
    sort_by: str = Query("created_at", description="Field to sort by"),
    order: str = Query("desc", description="asc or desc"),

    # Pagination
    skip: int = Query(0, ge=0, description="Records to skip"),
    limit: int = Query(20, ge=1, le=100, description="Max records"),

    # Search
    search: Optional[str] = Query(None, description="Search term"),

    db: Session = Depends(get_db),
    current_user: models.User = Depends(get_current_user)
):
    # Implementation below...

Query() parameters explained:

Query(None) - Optional, defaults to None
Query(0, ge=0) - Defaults to 0, must be >= 0
Query(20, ge=1, le=100) - Defaults to 20, must be 1-100

Why constrain limit to 100?

Without constraints:

GET /tasks?limit=1000000  # User requests 1 million records!
# Server crashes, database overloaded

With constraints:

limit: int = Query(20, ge=1, le=100)
GET /tasks?limit=1000000  # FastAPI returns 422 error
# Server protected!

Step 2: Building the Query

The magic of query chaining:

# Start with base query
query = db.query(models.Task).filter(models.Task.owner_id == current_user.id)

# Add filter if provided
if completed is not None:
    query = query.filter(models.Task.completed == completed)

# Add search if provided
if search:
    search_term = f"%{search}%"
    query = query.filter(
        or_(
            models.Task.title.ilike(search_term),
            models.Task.description.ilike(search_term)
        )
    )

# Add sorting
valid_sort_fields = ["created_at", "updated_at", "title", "completed"]
if sort_by not in valid_sort_fields:
    sort_by = "created_at"

column = getattr(models.Task, sort_by)

if order == "desc":
    query = query.order_by(desc(column))
else:
    query = query.order_by(asc(column))

# Add pagination
query = query.offset(skip).limit(limit)

# Execute query (only NOW does it hit database!)
return query.all()

Key insight: Query isn't executed until .all() is called!

This means:

query = db.query(Task)           # No SQL yet
query = query.filter(...)        # Still no SQL
query = query.order_by(...)      # Still building
tasks = query.all()              # NOW executes SQL!

Benefit: SQLAlchemy generates one optimized SQL query, not multiple.

Step 3: Understanding Filtering

Basic filtering:

if completed is not None:
    query = query.filter(models.Task.completed == completed)

Why is not None?

Consider these requests:

GET /tasks                    # completed = None (don't filter)
GET /tasks?completed=true     # completed = True (filter completed)
GET /tasks?completed=false    # completed = False (filter pending)

Without is not None:

# Bad
if completed:
    query = query.filter(Task.completed == completed)

# GET /tasks?completed=false
# completed = False
# if False: -> doesn't execute!
# Filter not applied! Bug!

With is not None:

# Good
if completed is not None:
    query = query.filter(Task.completed == completed)

# GET /tasks?completed=false
# completed = False
# if False is not None: -> True
# Filter applied correctly!

Step 4: Implementing Search

Case-insensitive search:

if search:
    search_term = f"%{search}%"
    query = query.filter(
        or_(
            models.Task.title.ilike(search_term),
            models.Task.description.ilike(search_term)
        )
    )

Breaking it down:

ILIKE (case-insensitive LIKE):

Task.title.ilike("%meeting%")

Matches:

"Team Meeting" ✅
"meeting notes" ✅
"MEETING" ✅
"Status meeting tomorrow" ✅

Doesn't match:

"Project discussion" ❌

The % wildcards:

meeting% - Starts with "meeting"
%meeting - Ends with "meeting"
%meeting% - Contains "meeting" anywhere

OR condition:

or_(
    Task.title.ilike("%meeting%"),
    Task.description.ilike("%meeting%")
)

SQL generated:

WHERE (title ILIKE '%meeting%' OR description ILIKE '%meeting%')

Searches both title AND description!

Step 5: Dynamic Sorting

valid_sort_fields = ["created_at", "updated_at", "title", "completed"]

if sort_by not in valid_sort_fields:
    sort_by = "created_at"  # Fallback to safe default

column = getattr(models.Task, sort_by)

if order == "desc":
    query = query.order_by(desc(column))
else:
    query = query.order_by(asc(column))

Why validate sort fields?

Without validation (DANGEROUS):

# User sends malicious request
GET /tasks?sort_by=hashed_password

column = getattr(Task, "hashed_password")  # Leaks password hashes!
query.order_by(column)

With validation (SAFE):

valid_fields = ["created_at", "title", "completed"]

if sort_by not in valid_fields:
    sort_by = "created_at"  # Use safe default

# User sends malicious request
GET /tasks?sort_by=hashed_password

# sort_by not in valid_fields
# Defaults to "created_at"
# Attack prevented!

Using getattr dynamically:

# Static (hardcoded)
if sort_by == "title":
    query.order_by(Task.title)
elif sort_by == "created_at":
    query.order_by(Task.created_at)
# ... many elif statements

# Dynamic (elegant)
column = getattr(Task, sort_by)
query.order_by(column)

Step 6: Pagination

query = query.offset(skip).limit(limit)

How pagination works:

# Page 1 (records 1-20)
skip=0, limit=20  # SELECT * FROM tasks LIMIT 20 OFFSET 0

# Page 2 (records 21-40)
skip=20, limit=20  # SELECT * FROM tasks LIMIT 20 OFFSET 20

# Page 3 (records 41-60)
skip=40, limit=20  # SELECT * FROM tasks LIMIT 20 OFFSET 40

Calculating skip:

page_number = 3
page_size = 20
skip = (page_number - 1) * page_size  # (3-1) * 20 = 40

Frontend implementation:

// React example
const [page, setPage] = useState(1);
const pageSize = 20;

fetch(`/tasks?skip=${(page - 1) * pageSize}&limit=${pageSize}`)

Step 7: Statistics Endpoint

@router.get("/stats", response_model=schemas.TaskStatsResponse)
def get_task_stats(
    db: Session = Depends(get_db),
    current_user: models.User = Depends(get_current_user)
):
    # Base query
    user_tasks = db.query(models.Task).filter(
        models.Task.owner_id == current_user.id
    )

    # Total count
    total = user_tasks.count()

    # Completed count
    completed = user_tasks.filter(models.Task.completed == True).count()

    # Pending count
    pending = total - completed

    # Completion rate (prevent division by zero!)
    completion_rate = (completed / total * 100) if total else 0

    # Tasks created today
    today_start = datetime.utcnow().replace(
        hour=0, minute=0, second=0, microsecond=0
    )
    created_today = user_tasks.filter(
        models.Task.created_at >= today_start
    ).count()

    # Tasks created this week
    week_start = today_start - timedelta(days=today_start.weekday())
    created_this_week = user_tasks.filter(
        models.Task.created_at >= week_start
    ).count()

    return {
        "total": total,
        "completed": completed,
        "pending": pending,
        "completion_rate": round(completion_rate, 2),
        "created_today": created_today,
        "created_this_week": created_this_week
    }

Understanding the date calculations:

Today start (midnight):

today_start = datetime.utcnow().replace(hour=0, minute=0, second=0)

# If now is: 2026-03-05 14:30:00
# today_start: 2026-03-05 00:00:00

Week start (Monday):

# weekday() returns 0=Monday, 6=Sunday
today_start.weekday()  # If Tuesday: returns 1

week_start = today_start - timedelta(days=1)
# Tuesday - 1 day = Monday

Edge case handling:

# Bad (crashes on empty task list)
completion_rate = (completed / total * 100)  # ZeroDivisionError!

# Good (handles zero total)
completion_rate = (completed / total * 100) if total else 0

Step 8: Database Indexing

Update models.py:

class Task(Base):
    __tablename__ = "tasks"

    # Add index=True to frequently queried columns
    title = Column(String(100), nullable=False, index=True)
    completed = Column(Boolean, default=False, index=True)
    created_at = Column(DateTime, default=datetime.utcnow, index=True)
    owner_id = Column(UUID(as_uuid=True), ForeignKey("users.id"), index=True)

    # Composite indexes for common query patterns
    __table_args__ = (
        Index('idx_owner_completed', 'owner_id', 'completed'),
        Index('idx_owner_created', 'owner_id', 'created_at'),
    )

Why indexes matter:

Without index:

SELECT * FROM tasks WHERE completed = false;
-- Database scans every row (full table scan)
-- 1,000 rows: ~50ms
-- 100,000 rows: ~5,000ms (5 seconds!)

With index:

SELECT * FROM tasks WHERE completed = false;
-- Database uses index (binary search)
-- 1,000 rows: ~5ms
-- 100,000 rows: ~50ms

~100x faster with indexes!

Composite indexes:

Index('idx_owner_completed', 'owner_id', 'completed')

Optimizes queries like:

WHERE owner_id = 'uuid' AND completed = false

Common in our API:

query.filter(Task.owner_id == current_user.id, Task.completed == False)

Index trade-offs:

Pros:

✅ Much faster queries
✅ Better user experience

Cons:

❌ Slightly slower writes (index must be updated)
❌ More disk space (~10-30% overhead)

Worth it for read-heavy applications!

Testing the Advanced Features

Test 1: Filtering

# Get pending tasks
curl -H "Authorization: Bearer TOKEN" \
  "http://localhost:8000/tasks?completed=false"

# Get completed tasks
curl -H "Authorization: Bearer TOKEN" \
  "http://localhost:8000/tasks?completed=true"

Test 2: Sorting

# Newest first
curl -H "Authorization: Bearer TOKEN" \
  "http://localhost:8000/tasks?sort_by=created_at&order=desc"

# Alphabetical by title
curl -H "Authorization: Bearer TOKEN" \
  "http://localhost:8000/tasks?sort_by=title&order=asc"

Test 3: Search

# Find tasks with "meeting"
curl -H "Authorization: Bearer TOKEN" \
  "http://localhost:8000/tasks?search=meeting"

# Case doesn't matter!
curl -H "Authorization: Bearer TOKEN" \
  "http://localhost:8000/tasks?search=MEETING"  # Same results

Test 4: Pagination

# First page
curl -H "Authorization: Bearer TOKEN" \
  "http://localhost:8000/tasks?skip=0&limit=20"

# Second page
curl -H "Authorization: Bearer TOKEN" \
  "http://localhost:8000/tasks?skip=20&limit=20"

Test 5: Combined Query

# Pending tasks with "project", sorted by date, page 1
curl -H "Authorization: Bearer TOKEN" \
  "http://localhost:8000/tasks?completed=false&search=project&sort_by=created_at&order=desc&limit=10"

Test 6: Statistics

curl -H "Authorization: Bearer TOKEN" \
  "http://localhost:8000/tasks/stats"

Response:

{
  "total": 25,
  "completed": 10,
  "pending": 15,
  "completion_rate": 40.0,
  "created_today": 3,
  "created_this_week": 8
}

Performance Comparison

Before (Day 3):

# Get all tasks (no pagination)
SELECT * FROM tasks WHERE owner_id = 'uuid'
# Returns: 1,000 tasks
# Time: 500ms
# Data transfer: 500KB

After (Day 4):

# Get filtered, sorted, paginated tasks
SELECT * FROM tasks 
WHERE owner_id = 'uuid' AND completed = false
ORDER BY created_at DESC
LIMIT 20 OFFSET 0
# Returns: 20 tasks
# Time: 50ms (with indexes)
# Data transfer: 10KB

Improvement: 10x faster, 50x less data!

Real-World Impact

Before advanced queries:

User has 5,000 tasks
Page loads all 5,000 (slow, crashes mobile apps)
Can't find specific tasks
No insights into productivity

After advanced queries:

Loads 20 tasks at a time (fast on any device)
Search finds tasks instantly
Filter shows only what's needed
Statistics provide insights

This transforms the API from proof-of-concept to product.

Common Mistakes I Avoided

Mistake 1: Not Checking completed is not None

# Bad - breaks for completed=false
if completed:
    query.filter(Task.completed == completed)

# Good - works for all cases
if completed is not None:
    query.filter(Task.completed == completed)

Mistake 2: No Sort Field Validation

# Bad - SQL injection risk
column = getattr(Task, sort_by)  # User could send "password"

# Good - whitelist validation
valid_fields = ["created_at", "title"]
if sort_by not in valid_fields:
    sort_by = "created_at"
column = getattr(Task, sort_by)

Mistake 3: Unlimited Pagination

# Bad - user can request millions of records
limit: int = Query(20)

# Good - constrain to reasonable max
limit: int = Query(20, ge=1, le=100)

Mistake 4: Case-Sensitive Search

# Bad - won't match "Meeting" when searching "meeting"
Task.title.like(f"%{search}%")

# Good - case-insensitive
Task.title.ilike(f"%{search}%")

Mistake 5: Division by Zero

# Bad - crashes on empty task list
completion_rate = (completed / total * 100)

# Good - handle zero case
completion_rate = (completed / total * 100) if total else 0

What I Learned

Technical Skills

✅ Advanced SQLAlchemy query composition

✅ FastAPI Query parameter validation

✅ Database indexing strategies

✅ Pagination implementation

✅ Case-insensitive search (ILIKE)

✅ Aggregation queries (COUNT, date filters)

✅ Dynamic column access with getattr

Conceptual Understanding

✅ Query chaining and lazy evaluation

✅ Index trade-offs (speed vs space)

✅ Security via input validation

✅ Edge case handling (zero division)

✅ Performance optimization strategies

API Design Patterns

✅ Standard query parameter conventions

✅ Pagination with skip/limit

✅ Multi-field sorting

✅ Combined query parameters

✅ Statistics endpoints for analytics

The "Aha!" Moments

1. Queries aren't executed until .all()
Building queries is cheap. Executing is expensive. Chain operations freely!

2. Indexes are like book indexes
Without index: Read every page to find "PostgreSQL"

With index: Jump directly to pages with "PostgreSQL"

3. Validation prevents attacks
Whitelisting sort fields isn't just best practice - it's security!

4. Query parameters compose beautifully
?completed=false&search=meeting&sort_by=created_at

Each parameter is independent but they work together!

Tomorrow's Plans

Day 5 will add:

Task categories/tags (many-to-many relationship)
Tag-based filtering
Bulk operations (delete multiple, update multiple)
Export functionality (CSV download)

But today? Today I celebrate building a production-ready query system that handles scale! 🎉

Resources That Helped

Full Code

Complete code on GitHub:
100-days-of-python/days/004

Time Investment: 2.5 hours

Knowledge Gained: Production query optimization skills

Feeling: Like I built something that actually scales 📊

Day 4/100 complete! ✅

Tomorrow: Many-to-many relationships and bulk operations

Following my journey?

📝 Blog | 🐦 Twitter | 💼 LinkedIn

Tags: #100DaysOfCode #Python #FastAPI #PostgreSQL #SQLAlchemy #BackendDevelopment #DatabaseOptimization #APIDesign #PerformanceTuning #LearningInPublic

Day 3/100: Adding JWT Authentication - Secure APIs with FastAPI

archi-jain — Fri, 06 Mar 2026 17:46:50 +0000

Part of my 100 Days of Code journey. Today we go from single-user to multi-user with proper security.

The Challenge

Add JWT-based authentication to the Task Management API, enabling user registration, login, and secure task isolation.

The Problem: Yesterday's API works great for one person. But what if multiple people use it? How do we know who owns which tasks? How do we keep data private?

The Solution: Implement industry-standard JWT authentication with password hashing, user management, and authorization.

Why Authentication Matters

Let me tell you what happens without authentication:

If you build APIs without authentication, you don't really have an application.

You have an open database with HTTP access.

Imagine a production task manager where:

User A could see User B's tasks.

User B could delete User A's tasks.

That would be chaos.

This isn't a feature. It's a security nightmare.

Today, we fix all of this.

What We're Building

Authentication has three parts:

Registration - Create a new user account
Login - Verify credentials, issue token
Authorization - Validate token, restrict access

Plus security features:

Password hashing (never store plain text!)
JWT tokens (stateless authentication)
User isolation (can't see others' data)

Understanding JWT

JWT (JSON Web Token) is how modern APIs handle authentication.

Traditional Session-Based Auth:

1. User logs in
2. Server creates session, stores in database
3. Server sends session ID to user
4. User sends session ID with each request
5. Server looks up session in database

Problem: Server must store sessions. Doesn't scale well.

JWT Token-Based Auth:

1. User logs in
2. Server creates JWT (signed with secret)
3. Server sends JWT to user
4. User sends JWT with each request
5. Server verifies signature (no database lookup!)

Benefit: Stateless! Server doesn't store anything.

JWT Structure

A JWT has three parts:

header.payload.signature

Example:
eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJhcmNoaSIsImV4cCI6MTY3ODkwMTIzNH0.rYVvXZ9f_Q...

Header (base64 encoded):

{
  "alg": "HS256",
  "typ": "JWT"
}

Payload (base64 encoded):

{
  "sub": "archi",
  "exp": 1678901234
}

Signature:

HMACSHA256(
  base64UrlEncode(header) + "." + base64UrlEncode(payload),
  SECRET_KEY
)

Key point: Anyone can decode header and payload (they're just base64). But only someone with the SECRET_KEY can create a valid signature.

This means:

✅ Server can verify token wasn't tampered with
✅ No database lookup needed
✅ Token is self-contained

Step-by-Step Implementation

Step 1: Installing Dependencies

pip install python-jose[cryptography] passlib[bcrypt] python-multipart

Why these packages?

python-jose: JWT creation and validation
passlib[bcrypt]: Password hashing (bcrypt algorithm)
python-multipart: Required for OAuth2 forms

Step 2: Environment Configuration

Create .env:

SECRET_KEY=your-super-secret-key-min-32-characters-long
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30

Generate a secure SECRET_KEY:

python3 -c "import secrets; print(secrets.token_urlsafe(32))"

CRITICAL: Never commit .env to Git. Add to .gitignore:

echo ".env" >> .gitignore

Step 3: Password Hashing

Why hash passwords?

Scenario: Your database gets hacked.

If you store plain text:

username: archi
password: securepass123  ← Attacker sees this!

If you hash:

username: archi
hashed_password: $2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/  ← Useless to attacker!

Create auth.py:

from passlib.context import CryptContext

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

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

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

How bcrypt works:

# Hashing (when user registers)
password = "securepass123"
hashed = get_password_hash(password)
# Result: $2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/...

# Verification (when user logs in)
entered_password = "securepass123"
is_valid = verify_password(entered_password, hashed)
# Result: True

# Wrong password
wrong_password = "wrongpass"
is_valid = verify_password(wrong_password, hashed)
# Result: False

Bcrypt automatically:

Generates random salt (prevents rainbow table attacks)
Uses slow algorithm (prevents brute force)
Includes salt in output (no need to store separately)

Step 4: JWT Token Creation

from datetime import datetime, timedelta
from jose import jwt
import os

SECRET_KEY = os.getenv("SECRET_KEY")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

def create_access_token(data: dict):
    to_encode = data.copy()

    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})

    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

    return encoded_jwt

Breaking it down:

# Input
data = {"sub": "archi"}

# Add expiration
to_encode = {"sub": "archi", "exp": 1678901234}

# Encode with secret
token = jwt.encode(to_encode, SECRET_KEY, algorithm="HS256")

# Result
"eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJhcmNoaSIsImV4cCI6MTY3ODkwMTIzNH0.signature"

What's "sub"?

JWT standard claim
"Subject" - who the token is about
We use username as the subject

Step 5: Token Validation

from jose import jwt, JWTError
from fastapi import HTTPException, status

def get_current_user(token: str, db: Session):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Invalid credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )

    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")

        if username is None:
            raise credentials_exception

    except JWTError:
        raise credentials_exception

    user = db.query(User).filter(User.username == username).first()

    if user is None:
        raise credentials_exception

    return user

What can go wrong?

Token expired → JWTError raised
Invalid signature → JWTError raised (token was tampered with)
Missing "sub" → credentials_exception
User doesn't exist → credentials_exception

All result in 401 Unauthorized - never tell attackers which part failed!

Step 6: OAuth2 Password Bearer

from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

What this does:

Tells FastAPI where to get tokens from
Automatically extracts token from Authorization: Bearer <token> header
Shows "Authorize" button in /docs
Returns 401 if token missing

Using it:

def get_current_user(
    token: str = Depends(oauth2_scheme),  # Auto-extracts token!
    db: Session = Depends(get_db)
):
    # token is already extracted from Authorization header
    # ...

Step 7: User Model

Update models.py:

from sqlalchemy import Column, String, Boolean, ForeignKey
from sqlalchemy.orm import relationship

class User(Base):
    __tablename__ = "users"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    username = Column(String(50), unique=True, nullable=False, index=True)
    email = Column(String(100), unique=True, nullable=False, index=True)
    hashed_password = Column(String(255), nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)

    # Relationship
    tasks = relationship("Task", back_populates="owner", cascade="all, delete-orphan")

Key fields:

username: Must be unique (for login)
email: Must be unique (for verification)
hashed_password: NEVER plain text!
is_active: Allows disabling accounts
index=True: Fast lookups (we search by username often)

Relationship:

tasks = relationship("Task", back_populates="owner", cascade="all, delete-orphan")

User can have many tasks
If user deleted, delete their tasks too (cascade)

Step 8: Updated Task Model

class Task(Base):
    __tablename__ = "tasks"

    # ... existing fields ...

    # NEW: Foreign key
    owner_id = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False)

    # NEW: Relationship
    owner = relationship("User", back_populates="tasks")

Foreign key:

owner_id = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False)

Links task to a user
ondelete="CASCADE": If user deleted, delete task
nullable=False: Every task MUST have an owner

Database visualization:

users table:
id | username | email | hashed_password
1  | archi    | ...   | $2b$12...

tasks table:
id | title        | owner_id (FK)
1  | Learn JWT    | 1 (points to archi)
2  | Build API    | 1 (points to archi)

Step 9: Pydantic Schemas

Update schemas.py:

from pydantic import BaseModel, EmailStr

class UserCreate(BaseModel):
    username: str
    email: EmailStr
    password: str

class UserResponse(BaseModel):
    id: UUID
    username: str
    email: str
    is_active: bool
    created_at: datetime
    # NO PASSWORD! Never expose hashes

    class Config:
        from_attributes = True

class Token(BaseModel):
    access_token: str
    token_type: str

Critical: UserResponse excludes password!

# BAD - exposes password hash
class UserResponse(BaseModel):
    username: str
    hashed_password: str  # ❌ NEVER!

# GOOD
class UserResponse(BaseModel):
    username: str
    # No password field ✅

Step 10: Registration Endpoint

Create routers/auth_router.py:

@router.post("/register", response_model=schemas.UserResponse)
def register(user: schemas.UserCreate, db: Session = Depends(get_db)):
    # Check if username exists
    existing = db.query(models.User).filter(
        models.User.username == user.username
    ).first()

    if existing:
        raise HTTPException(status_code=400, detail="Username already exists")

    # Create user with hashed password
    new_user = models.User(
        username=user.username,
        email=user.email,
        hashed_password=get_password_hash(user.password)  # Hash it!
    )

    db.add(new_user)
    db.commit()
    db.refresh(new_user)

    return new_user

Security checks:

Verify username doesn't exist - Prevents duplicates
Hash password before storing - NEVER store plain text!
Return user without password - UserResponse excludes it

Step 11: Login Endpoint

@router.post("/login", response_model=schemas.Token)
def login(credentials: schemas.UserLogin, db: Session = Depends(get_db)):
    # Find user
    user = db.query(models.User).filter(
        models.User.username == credentials.username
    ).first()

    # Verify user exists AND password correct
    if not user or not verify_password(credentials.password, user.hashed_password):
        raise HTTPException(status_code=401, detail="Invalid credentials")

    # Create token
    token = create_access_token({"sub": user.username})

    return {"access_token": token, "token_type": "bearer"}

Security best practices:

1. Generic error messages:

# BAD - tells attacker what's wrong
if not user:
    raise HTTPException(detail="Username not found")
if not verify_password(...):
    raise HTTPException(detail="Wrong password")

# GOOD - same error for both
if not user or not verify_password(...):
    raise HTTPException(detail="Invalid credentials")

2. Always verify password:

# Check both conditions together
if not user or not verify_password(credentials.password, user.hashed_password):

This prevents timing attacks (where attacker can tell if username exists based on response time).

Step 12: Protected Routes

Update routers/tasks_router.py:

@router.post("/", response_model=schemas.TaskResponse)
def create_task(
    task: schemas.TaskCreate,
    db: Session = Depends(get_db),
    current_user: models.User = Depends(get_current_user)  # 🔒 Requires auth!
):
    new_task = models.Task(
        title=task.title,
        description=task.description,
        owner_id=current_user.id  # Assign to current user
    )

    db.add(new_task)
    db.commit()
    db.refresh(new_task)

    return new_task

What Depends(get_current_user) does:

Extract token from Authorization header
Decode and validate token
Look up user in database
Return user object
If any step fails → 401 Unauthorized

User isolation:

@router.get("/", response_model=List[schemas.TaskResponse])
def get_tasks(
    db: Session = Depends(get_db),
    current_user: models.User = Depends(get_current_user)
):
    # Only return current user's tasks!
    tasks = db.query(models.Task).filter(
        models.Task.owner_id == current_user.id
    ).all()

    return tasks

SQL equivalent:

SELECT * FROM tasks WHERE owner_id = current_user.id;

Why this matters:

Alice (user id=1) calls GET /tasks
Query: WHERE owner_id = 1
Only Alice's tasks returned
Bob (user id=2) calls GET /tasks
Query: WHERE owner_id = 2
Only Bob's tasks returned

Perfect isolation!

Step 13: Complete Flow Example

User journey:

1. Register:

POST /auth/register
{
  "username": "archi",
  "email": "archi@example.com",
  "password": "securepass123"
}

Response:
{
  "id": "uuid-here",
  "username": "archi",
  "email": "archi@example.com",
  "is_active": true,
  "created_at": "2026-03-05T..."
}

2. Login:

POST /auth/login
{
  "username": "archi",
  "password": "securepass123"
}

Response:
{
  "access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
  "token_type": "bearer"
}

3. Create task (with token):

POST /tasks/
Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...
{
  "title": "My Task",
  "description": "Only I can see this"
}

Response:
{
  "id": "task-uuid",
  "title": "My Task",
  "owner_id": "archi-uuid",  ← Automatically set!
  ...
}

4. Try without token (fails):

POST /tasks/
{
  "title": "Unauthorized"
}

Response: 401 Unauthorized
{
  "detail": "Not authenticated"
}

Testing the Authentication

Using cURL

Register:

curl -X POST "http://localhost:8000/auth/register" \
  -H "Content-Type: application/json" \
  -d '{"username":"archi","email":"archi@example.com","password":"test1234"}'

Login:

curl -X POST "http://localhost:8000/auth/login" \
  -H "Content-Type: application/json" \
  -d '{"username":"archi","password":"test1234"}'

Copy the access_token from response.

Create task with token:

curl -X POST "http://localhost:8000/tasks/" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_TOKEN_HERE" \
  -d '{"title":"Authenticated Task","description":"With JWT"}'

Using Interactive Docs

Go to http://localhost:8000/docs
Click "Authorize" button (🔓 icon)
Click "POST /auth/login"
Enter username and password
Click "Try it out" → "Execute"
Copy the access_token
Click "Authorize" button again
Paste token (will auto-add "Bearer ")
Click "Authorize"
Now all endpoints show 🔒 (locked/authorized)
Try creating tasks - they'll be yours!

Security Considerations

What We Did Right

✅ Password hashing - Never store plain text

✅ Token expiration - Tokens expire after 30 min

✅ Generic errors - Don't leak info to attackers

✅ User isolation - Can't access others' data

✅ Environment variables - Secrets not in code

What Could Be Improved (Future)

1. Refresh tokens:

Access token expires quickly (30 min)
Refresh token lasts longer (7 days)
Use refresh to get new access token

2. Password strength validation:

import re

def validate_password(password: str):
    if len(password) < 8:
        raise ValueError("Too short")
    if not re.search(r"[A-Z]", password):
        raise ValueError("Need uppercase")
    if not re.search(r"[0-9]", password):
        raise ValueError("Need number")

3. Rate limiting:

Prevent brute force attacks
Limit login attempts per IP
Lock account after failures

4. Email verification:

Send confirmation email
User must verify before login

5. Two-factor authentication (2FA):

SMS or authenticator app
Extra security layer

Common Mistakes I Avoided

Mistake 1: Storing Plain Text Passwords

# NEVER DO THIS ❌
new_user = User(
    username=username,
    password=password  # Plain text!
)

# ALWAYS DO THIS ✅
new_user = User(
    username=username,
    hashed_password=get_password_hash(password)
)

Mistake 2: Exposing Passwords in API

# BAD ❌
class UserResponse(BaseModel):
    username: str
    hashed_password: str  # Exposed!

# GOOD ✅
class UserResponse(BaseModel):
    username: str
    # No password field!

Mistake 3: Not Checking Task Ownership

# BAD - Anyone can modify any task ❌
task = db.query(Task).filter(Task.id == task_id).first()

# GOOD - Only owner can modify ✅
task = db.query(Task).filter(
    Task.id == task_id,
    Task.owner_id == current_user.id
).first()

Mistake 4: Hardcoding SECRET_KEY

# BAD ❌
SECRET_KEY = "my-secret-123"  # In code!

# GOOD ✅
SECRET_KEY = os.getenv("SECRET_KEY")  # From .env

What I Learned

Technical Skills

✅ JWT token creation and validation

✅ Password hashing with bcrypt

✅ OAuth2 bearer token authentication

✅ FastAPI security dependencies

✅ SQLAlchemy relationships (one-to-many)

✅ User isolation and authorization

Conceptual Understanding

✅ Authentication vs Authorization

✅ Stateless vs Stateful sessions

✅ Token-based auth advantages

✅ Why never store plain text passwords

✅ Salt and hashing concepts

✅ Security best practices

Real-World Patterns

✅ Multi-user application architecture

✅ Router organization and modularity

✅ Environment-based configuration

✅ Error handling that doesn't leak info

The "Aha!" Moments

1. JWT is self-contained
No database lookup needed to validate! The signature proves authenticity.

2. Bcrypt handles salt automatically
Don't need to generate/store salt separately. It's in the hash!

3. Dependency injection is powerful

current_user: User = Depends(get_current_user)

One line gets authenticated user in every route!

4. Relationships simplify queries
Instead of manual joins, just user.tasks or task.owner.

Next Steps

Next improvements I want to add:

Refresh tokens
Pagination for tasks
Password strength validation
Email verification
Role-based authorization

But today? Today I celebrate building a secure, multi-user API that could actually go to production! 🎉

Resources That Helped

Full Code

Complete code on GitHub:
100-days-of-python/days/003

Time Investment: 3 hours

Knowledge Gained: Production authentication skills

Feeling: Like I can build secure systems 🔐

Day 3/100 complete! ✅

Tomorrow: Advanced features (filtering, search, pagination)

Following my journey?

📝 Blog | 🐦 Twitter | 💼 LinkedIn

Tags: #100DaysOfCode #Python #FastAPI #JWT #Authentication #Security #BackendDevelopment #OAuth2 #PasswordHashing #LearningInPublic

Day 2/100: From In-Memory to PostgreSQL - Database Integration with SQLAlchemy

archi-jain — Thu, 05 Mar 2026 08:28:49 +0000

Part of my 100 Days of Code journey. Yesterday I built a REST API. Today I make it real.

The Challenge

Take the Task Management API from Day 1 and upgrade it from in-memory storage to a production-ready PostgreSQL database using SQLAlchemy ORM.

The Problem: In-memory storage disappears when the server restarts. Not exactly production-ready.

The Solution: Integrate PostgreSQL with proper ORM patterns, session management, and environment configuration.

Why Databases Matter

Let me be blunt: your API is useless without persistent storage.

Day 1's in-memory dictionary was perfect for learning HTTP and REST patterns. But the moment you restart your server, poof—all data gone. That's not an API. That's a expensive notepad.

Today, we fix that.

What We're Building

We're keeping all the functionality from Day 1:

Create, read, update, delete tasks
Mark tasks complete/incomplete
Proper error handling

But now with:

✅ Data that survives server restarts
✅ Scalable database architecture
✅ Production-ready patterns
✅ Environment-based configuration

Technology Stack

Database: PostgreSQL 15

ORM: SQLAlchemy

Environment Config: python-dotenv

Database Driver: psycopg2

Step-by-Step Implementation

Step 1: Installing PostgreSQL

I used Docker because it's the easiest way to get PostgreSQL running without cluttering my system:

docker run --name postgres-tasks \
  -e POSTGRES_USER=postgres \
  -e POSTGRES_PASSWORD=postgres \
  -e POSTGRES_DB=tasks_db \
  -p 5432:5432 \
  -d postgres:15

What this does:

Creates a PostgreSQL container named postgres-tasks
Sets up a database called tasks_db
Exposes it on port 5432 (PostgreSQL default)
Runs in detached mode (background)

Verify it's running:

docker ps

You should see your postgres-tasks container running.

Step 2: Installing Python Dependencies

pip install sqlalchemy psycopg2-binary python-dotenv

Package purposes:

sqlalchemy: The ORM that converts Python to SQL
psycopg2-binary: PostgreSQL adapter (connects Python to PostgreSQL)
python-dotenv: Loads environment variables from .env files

Step 3: Project Structure

days/002/
├── database.py      # Database connection setup
├── models.py        # SQLAlchemy ORM models
├── schemas.py       # Pydantic models (API validation)
├── main.py          # FastAPI application
├── .env             # Environment variables (DON'T COMMIT!)
└── requirements.txt

Step 4: Environment Configuration

Create .env file:

DATABASE_URL=postgresql://postgres:postgres@localhost:5432/tasks_db

Breaking down the URL:

postgresql:// - Database type
postgres:postgres - username:password
@localhost:5432 - host:port
/tasks_db - database name

Security Note: Never commit .env files! Add to .gitignore:

echo ".env" >> .gitignore

Step 5: Database Connection Setup

Create database.py:

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import os
from dotenv import load_dotenv

load_dotenv()

DATABASE_URL = os.getenv("DATABASE_URL")

engine = create_engine(
    DATABASE_URL,
    pool_pre_ping=True,
    echo=True
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Understanding each part:

create_engine:

Creates a connection pool to the database
pool_pre_ping=True: Checks connections before using (prevents stale connections)
echo=True: Logs all SQL queries (great for learning, remove in production)

SessionLocal:

Factory for creating database sessions
A session is like a workspace for database operations
autocommit=False: We control when to save changes
autoflush=False: We control when to send changes to database

Base:

Base class that all our models inherit from
SQLAlchemy uses this to track all models

get_db() function:

FastAPI dependency that provides database sessions
yield gives the session to the route
finally ensures session always closes (even on errors)

This pattern is crucial - it prevents database connection leaks.

Step 6: Creating the Database Model

Create models.py:

from sqlalchemy import Column, String, Boolean, DateTime
from sqlalchemy.dialects.postgresql import UUID
from datetime import datetime
import uuid
from database import Base

class Task(Base):
    __tablename__ = "tasks"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    title = Column(String(100), nullable=False)
    description = Column(String(500), nullable=True)
    completed = Column(Boolean, default=False, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False)

    def __repr__(self):
        return f"<Task(id={self.id}, title={self.title})>"

Deep dive into the model:

tablename = "tasks":

Tells SQLAlchemy what to name the database table

id Column:

Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)

UUID(as_uuid=True): Uses PostgreSQL's native UUID type
primary_key=True: This column uniquely identifies each row
default=uuid.uuid4: Auto-generates UUID when creating new tasks

Why UUIDs over integers?

Globally unique (can merge databases without ID conflicts)
Hard to guess (security)
Can generate client-side
Better for distributed systems

title and description:

Column(String(100), nullable=False)
Column(String(500), nullable=True)

String(100): Maximum 100 characters
nullable=False: Required field
nullable=True: Optional field

completed:

Column(Boolean, default=False, nullable=False)

Boolean type (true/false)
Defaults to False (new tasks are incomplete)
Cannot be null

Timestamps:

created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False)

default=datetime.utcnow: Sets timestamp when task created
onupdate=datetime.utcnow: Auto-updates timestamp on any modification
Always use UTC in databases! Convert to local time in the frontend

repr method:

Defines how the object appears when printed
Useful for debugging
Example: <Task(id=uuid, title="Learn FastAPI")>

Step 7: Pydantic Schemas

Create schemas.py (similar to Day 1, but with one important change):

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

class TaskCreate(BaseModel):
    title: str = Field(..., max_length=100)
    description: Optional[str] = Field(None, max_length=500)

class TaskUpdate(BaseModel):
    title: Optional[str] = Field(None, max_length=100)
    description: Optional[str] = Field(None, max_length=500)

class TaskResponse(BaseModel):
    id: UUID
    title: str
    description: Optional[str]
    completed: bool
    created_at: datetime
    updated_at: datetime

    class Config:
        from_attributes = True

The critical addition:

class Config:
    from_attributes = True

This tells Pydantic to convert SQLAlchemy model objects to Pydantic models.

Why do we need both models.py and schemas.py?

models.py (SQLAlchemy): Defines database structure
schemas.py (Pydantic): Defines API structure

They're similar but serve different purposes:

SQLAlchemy talks to the database
Pydantic validates API requests/responses

Step 8: Creating the FastAPI Application

Create main.py:

from fastapi import FastAPI, HTTPException, status, Depends
from sqlalchemy.orm import Session
from typing import List
from uuid import UUID
import models
import schemas
from database import engine, get_db

# Create all tables
models.Base.metadata.create_all(bind=engine)

app = FastAPI(
    title="Task Management API with PostgreSQL",
    version="2.0.0"
)

@app.post("/tasks", response_model=schemas.TaskResponse, status_code=status.HTTP_201_CREATED)
def create_task(task: schemas.TaskCreate, db: Session = Depends(get_db)):
    new_task = models.Task(
        title=task.title,
        description=task.description
    )
    db.add(new_task)
    db.commit()
    db.refresh(new_task)
    return new_task

Let's break down the create endpoint:

1. Function signature:

def create_task(task: schemas.TaskCreate, db: Session = Depends(get_db)):

task: schemas.TaskCreate: FastAPI auto-validates request body
db: Session = Depends(get_db): FastAPI injects database session

2. Creating the model instance:

new_task = models.Task(
    title=task.title,
    description=task.description
)

Creates SQLAlchemy Task object
Note: We DON'T set id, created_at, updated_at, completed
Those are auto-generated by the database!

3. Saving to database:

db.add(new_task)
db.commit()
db.refresh(new_task)

db.add(): Stages the task for insertion
db.commit(): Actually saves to database
db.refresh(): Fetches auto-generated values (id, timestamps)

Why three separate calls?

Allows batching multiple operations
Can rollback before commit if needed
Refresh gets database-generated values

4. Returning the task:

return new_task

FastAPI automatically:

Converts SQLAlchemy model to Pydantic schema
Serializes to JSON
Sets proper Content-Type header

The Other Endpoints

Get All Tasks:

@app.get("/tasks", response_model=List[schemas.TaskResponse])
def get_tasks(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    tasks = db.query(models.Task).offset(skip).limit(limit).all()
    return tasks

Understanding the query:

db.query(models.Task): Start building a query
.offset(skip): Skip first N records (pagination)
.limit(limit): Return max N records
.all(): Execute query, return all results

SQL equivalent:

SELECT * FROM tasks OFFSET 0 LIMIT 100;

Get Single Task:

@app.get("/tasks/{task_id}", response_model=schemas.TaskResponse)
def get_task(task_id: UUID, db: Session = Depends(get_db)):
    task = db.query(models.Task).filter(models.Task.id == task_id).first()

    if not task:
        raise HTTPException(status_code=404, detail="Task not found")

    return task

Query breakdown:

.filter(models.Task.id == task_id): WHERE clause
.first(): Returns first result or None

SQL equivalent:

SELECT * FROM tasks WHERE id = 'some-uuid' LIMIT 1;

Update Task:

@app.put("/tasks/{task_id}", response_model=schemas.TaskResponse)
def update_task(task_id: UUID, task_update: schemas.TaskUpdate, db: Session = Depends(get_db)):
    task = db.query(models.Task).filter(models.Task.id == task_id).first()

    if not task:
        raise HTTPException(status_code=404, detail="Task not found")

    if task_update.title is not None:
        task.title = task_update.title

    if task_update.description is not None:
        task.description = task_update.description

    db.commit()
    db.refresh(task)
    return task

Why check is not None?

Consider this request:

{
  "title": "New Title"
  // description not sent
}

We want to update title but keep description unchanged.

if task_update.title is not None:  # True
    task.title = task_update.title

if task_update.description is not None:  # False
    task.description = task_update.description  # Skipped!

This enables partial updates!

Delete Task:

@app.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_task(task_id: UUID, db: Session = Depends(get_db)):
    task = db.query(models.Task).filter(models.Task.id == task_id).first()

    if not task:
        raise HTTPException(status_code=404, detail="Task not found")

    db.delete(task)
    db.commit()
    return None

Why return None for 204?

HTTP 204 means "success, no content"
Returning None tells FastAPI not to send a response body
Still returns 204 status code

Testing the API

Start the server:

uvicorn main:app --reload

Create a task:

curl -X POST "http://localhost:8000/tasks" \
  -H "Content-Type: application/json" \
  -d '{"title": "Learn SQLAlchemy", "description": "Integrate PostgreSQL with FastAPI"}'

The magic moment:

Stop the server (Ctrl+C)
Start it again
Get all tasks:

curl http://localhost:8000/tasks

The task is still there! 🎉

That's the power of persistent storage.

Checking the Database Directly

docker exec -it postgres-tasks psql -U postgres -d tasks_db

SELECT * FROM tasks;

You'll see your task with all columns:

id (UUID)
title
description
completed (false)
created_at (timestamp)
updated_at (timestamp)

Key Concepts Learned

ORM (Object-Relational Mapping)

Without ORM (raw SQL):

cursor.execute("INSERT INTO tasks (title, description) VALUES (%s, %s)", (title, desc))

With ORM:

new_task = Task(title=title, description=desc)
db.add(new_task)
db.commit()

Benefits:

Type-safe (Python catches errors before database does)
Prevents SQL injection automatically
Database-agnostic (can switch from PostgreSQL to MySQL)
Easier to maintain and test

Database Sessions

Think of a session as a transaction workspace:

db = SessionLocal()  # Open workspace
try:
    db.add(task1)
    db.add(task2)
    db.commit()  # Save all changes
except:
    db.rollback()  # Undo all changes
finally:
    db.close()  # Clean up

FastAPI's Depends(get_db) handles this automatically!

Connection Pooling

SQLAlchemy maintains a pool of database connections:

Reuses connections (faster than creating new ones)
Limits max connections (prevents overwhelming database)
Auto-reconnects if connection dies

All handled automatically by create_engine().

Common Mistakes and How I Avoided Them

Mistake 1: Not Closing Sessions

Wrong:

db = SessionLocal()
db.add(task)
db.commit()
# Forgot to close!

Right (with Depends):

def create_task(db: Session = Depends(get_db)):
    # Session auto-closes when function ends

Mistake 2: Forgetting to Commit

Wrong:

db.add(task)
return task  # Task not saved!

Right:

db.add(task)
db.commit()  # Actually save
db.refresh(task)  # Get auto-generated fields
return task

Mistake 3: Hardcoding Database URL

Wrong:

DATABASE_URL = "postgresql://user:pass@localhost/db"  # In code!

Right:

DATABASE_URL = os.getenv("DATABASE_URL")  # From .env

Never commit credentials!

What I Learned Today

Technical Skills:

✅ PostgreSQL setup and configuration

✅ SQLAlchemy ORM models and relationships

✅ Database session lifecycle management

✅ Environment-based configuration

✅ Query building with SQLAlchemy

✅ UUID vs integer primary keys

✅ Timestamp auto-management

Conceptual Understanding:

✅ Why ORMs exist and when to use them

✅ Database connection pooling

✅ Transaction management

✅ The difference between staging and committing

✅ Why sessions must be closed

Production Patterns:

✅ Environment variable configuration

✅ Dependency injection for resources

✅ Proper error handling

✅ Security best practices (no hardcoded credentials)

The "Aha!" Moments

1. Sessions are NOT connections

A session is a workspace for database operations
Connections are managed by the engine's pool
One session can use multiple connections

2. ORM saves so much boilerplate

No manual SQL string building
No manual parameter binding
Type checking at Python level

3. Dependency injection is elegant

def my_route(db: Session = Depends(get_db)):

That one line:

Gets a database session
Provides it to the function
Ensures it's closed afterward
Handles errors automatically

Beautiful.

Next Steps

Tomorrow (Day 3), I'm planning to add:

Advanced query features (filtering, searching)
User model with relationships
Authentication basics

But today? Today I celebrate having a real, production-ready database powering my API. 🎉

Resources That Helped

Full Code

Complete code available on GitHub:
100-days-of-python/days/002

Time Investment: 2.5 hours

Knowledge Gained: Production database integration skills

Feeling: Like a real backend developer 💪

Day 2/100 complete! ✅

Tomorrow: Advanced queries and relationships

Following my 100 Days journey?

📝 Blog | 🐦 Twitter | 💼 LinkedIn

Tags: #100DaysOfCode #Python #FastAPI #PostgreSQL #SQLAlchemy #BackendDevelopment #DatabaseDesign #LearningInPublic

Day 1/100: Building Your First FastAPI REST API - A Complete Beginner's Guide

archi-jain — Wed, 04 Mar 2026 16:48:45 +0000

Day 1/100: Building Your First FastAPI REST API - A Complete Beginner's Guide

Part of my 100 Days of Code journey. Follow along as I build production-ready Python skills.

Introduction

Welcome to Day 1 of my 100 Days of Code challenge! Today, I'm diving into FastAPI, a modern Python web framework that's taking the backend world by storm. If you've been using Flask or Django and wondering what all the FastAPI hype is about, this post is for you.

By the end of this tutorial, you'll understand how to build a complete REST API from scratch, including proper validation, error handling, and auto-generated documentation.

Why FastAPI?

Before we dive into code, let's talk about why FastAPI matters:

Fast: As the name suggests, it's one of the fastest Python frameworks available (comparable to Node.js and Go)
Modern: Built on modern Python features like type hints and async/await
Auto-validation: Pydantic models validate requests automatically
Auto-docs: Interactive API documentation generated for free
Industry adoption: Used by Uber, Netflix, Microsoft, and other tech giants

The Challenge

Build a Task Management API with these features:

Create, read, update, and delete tasks (CRUD)
Mark tasks as complete/incomplete
Proper error handling
Request/response validation
Auto-generated API documentation

Project Setup

Installing Dependencies

First, create a virtual environment and install FastAPI:

# Create virtual environment
python3 -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install dependencies
pip install fastapi uvicorn

What are these packages?

fastapi: The web framework itself
uvicorn: An ASGI server to run our FastAPI application

Project Structure

Create a single file called main.py. For this beginner project, we'll keep everything in one file for simplicity.

Building the API - Step by Step

Step 1: Initial Setup and Imports

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

Line-by-line explanation:

FastAPI: The main class to create our application
HTTPException: Used to return HTTP error responses (like 404 Not Found)
status: Contains HTTP status code constants (201, 404, etc.)
BaseModel from Pydantic: Base class for our data models
Field: Adds validation constraints to model fields
Optional: Type hint for fields that can be None
Dict, List: Type hints for dictionaries and lists
datetime: For timestamps
uuid: For generating unique task IDs

Step 2: Initialize the Application

app = FastAPI(
    title="Task Management API",
    description="A simple REST API for managing tasks",
    version="1.0.0"
)

What's happening here?

We create a FastAPI instance with metadata. This information appears in the auto-generated documentation at /docs.

Step 3: Create In-Memory Storage

tasks_db: Dict[str, dict] = {}

Why a dictionary?

For Day 1, we're using a simple Python dictionary to store tasks. The keys are task IDs (strings), and the values are task dictionaries. This simulates a database without the complexity.

Type hint breakdown:

Dict[str, dict]: A dictionary with string keys and dict values
: Dict[str, dict] = {}: Type annotation with default empty dict

In production, you'd use a real database (PostgreSQL, MongoDB, etc.), but this is perfect for learning.

Step 4: Define Data Models with Pydantic

TaskCreate Model

class TaskCreate(BaseModel):
    title: str = Field(..., max_length=100, description="Task title")
    description: Optional[str] = Field(None, max_length=500, description="Task description")

Breaking it down:

class TaskCreate(BaseModel): Inherits from Pydantic's BaseModel
title: str: Required string field
Field(...): The ... means this field is required
max_length=100: Validation - title can't exceed 100 characters
description="Task title": Shows in API docs
Optional[str]: This field can be a string or None
Field(None, ...): Default value is None (field is optional)

Why separate models?

We create different models for different purposes:

TaskCreate: What the user sends when creating a task
TaskUpdate: What can be updated
TaskResponse: What the API returns (includes ID, timestamps)

This separation follows the Single Responsibility Principle.

TaskUpdate Model

class TaskUpdate(BaseModel):
    title: Optional[str] = Field(None, max_length=100)
    description: Optional[str] = Field(None, max_length=500)

Why everything is Optional here?

When updating, users might want to change only the title OR only the description, not necessarily both. Making fields optional allows partial updates.

TaskResponse Model

class TaskResponse(BaseModel):
    id: str
    title: str
    description: Optional[str]
    completed: bool
    created_at: datetime
    updated_at: datetime

What's different?

This model includes fields that the API generates automatically:

id: Auto-generated UUID
completed: Default False, set by the system
created_at: Timestamp when task was created
updated_at: Timestamp when task was last modified

Users don't send these fields; the API creates them.

Step 5: Helper Function for Error Handling

def get_task_or_404(task_id: str):
    task = tasks_db.get(task_id)

    if not task:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found"
        )

    return task

Why create this helper?

Multiple endpoints need to check if a task exists. Instead of repeating this code, we create a reusable function. This follows the DRY principle (Don't Repeat Yourself).

How it works:

tasks_db.get(task_id): Safely get task (returns None if not found)
If task doesn't exist, raise HTTPException with 404 status
If task exists, return it

HTTPException explained:

When raised, FastAPI automatically converts this to a proper HTTP error response:

{
  "detail": "Task not found"
}

Step 6: Create Task Endpoint

@app.post("/tasks", response_model=TaskResponse, status_code=status.HTTP_201_CREATED)
async def create_task(task: TaskCreate):
    task_id = str(uuid.uuid4())
    now = datetime.utcnow()

    new_task = {
        "id": task_id,
        "title": task.title,
        "description": task.description,
        "completed": False,
        "created_at": now,
        "updated_at": now
    }

    tasks_db[task_id] = new_task

    return new_task

Decorator breakdown:

@app.post("/tasks"): This function handles POST requests to /tasks
response_model=TaskResponse: FastAPI validates the response matches this model
status_code=status.HTTP_201_CREATED: Returns 201 (resource created) instead of default 200

Why async def?

FastAPI is built for asynchronous programming. Even though this function doesn't use await, using async def is best practice and allows FastAPI to handle it efficiently.

Function parameters:

task: TaskCreate: FastAPI automatically:
1. Parses the JSON request body
2. Validates it against TaskCreate model
3. Converts it to a TaskCreate object
4. If validation fails, returns 422 error automatically

Step-by-step logic:

Generate unique ID: str(uuid.uuid4()) creates a random UUID like "a3bb189e-8bf9-..."
Get current timestamp: datetime.utcnow() gets UTC time (best practice for APIs)
Create task dictionary with all required fields
Store in our "database": tasks_db[task_id] = new_task
Return the task (FastAPI automatically converts to JSON)

Step 7: Get All Tasks Endpoint

@app.get("/tasks", response_model=List[TaskResponse])
async def get_all_tasks():
    return list(tasks_db.values())

Simple but powerful!

@app.get("/tasks"): Handles GET requests to /tasks
response_model=List[TaskResponse]: Returns a list of TaskResponse objects
list(tasks_db.values()): Gets all task dictionaries from our storage

What FastAPI does automatically:

Validates each task in the list matches TaskResponse
Converts Python datetime objects to ISO format strings
Converts the entire list to JSON
Sets proper Content-Type header

Step 8: Get Single Task Endpoint

@app.get("/tasks/{task_id}", response_model=TaskResponse)
async def get_task(task_id: str):
    task = get_task_or_404(task_id)
    return task

Path parameters explained:

"/tasks/{task_id}": Curly braces define a path parameter
task_id: str: FastAPI extracts this from the URL
Example: GET /tasks/abc123 → task_id = "abc123"

Flow:

FastAPI extracts task_id from URL
Our helper checks if it exists
If not found, raises 404 automatically
If found, returns the task

Step 9: Update Task Endpoint

@app.put("/tasks/{task_id}", response_model=TaskResponse)
async def update_task(task_id: str, task_update: TaskUpdate):
    task = get_task_or_404(task_id)

    if task_update.title is not None:
        task["title"] = task_update.title

    if task_update.description is not None:
        task["description"] = task_update.description

    task["updated_at"] = datetime.utcnow()

    return task

PUT vs PATCH:

PUT: Update entire resource (we're using this)
PATCH: Partial update (we'll use this for completion toggle)

Why check is not None?

Fields in TaskUpdate are optional. We only update fields that were actually sent in the request:

# Request: {"title": "New Title"}
# Only title is updated, description stays the same

if task_update.title is not None:  # True
    task["title"] = task_update.title

if task_update.description is not None:  # False (not sent)
    task["description"] = task_update.description  # Skipped!

Always update timestamp:

Even if only one field changed, we update updated_at to track when the task was last modified.

Step 10: Toggle Completion Endpoint

@app.patch("/tasks/{task_id}/complete", response_model=TaskResponse)
async def toggle_task_completion(task_id: str):
    task = get_task_or_404(task_id)

    task["completed"] = not task["completed"]
    task["updated_at"] = datetime.utcnow()

    return task

Why PATCH here?

PATCH is perfect for small, specific updates. This endpoint does one thing: toggle completion status.

The toggle logic:

task["completed"] = not task["completed"]

If completed is True → becomes False
If completed is False → becomes True

Simple and elegant!

Step 11: Delete Task Endpoint

@app.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_task(task_id: str):
    task = get_task_or_404(task_id)
    del tasks_db[task_id]
    return None

HTTP 204 No Content:

Standard for successful deletions. Means:

✅ Request succeeded
ℹ️ No content in response body

Why check if exists first?

We want to return 404 if the task doesn't exist, not silently succeed. This gives clients better feedback.

Delete operation:

del tasks_db[task_id]

Removes the key-value pair from our dictionary.

Testing Your API

Run the Server

uvicorn main:app --reload

Command breakdown:

uvicorn: The ASGI server
main:app: File name (main.py) : FastAPI instance (app)
--reload: Auto-restart on code changes (development only!)

Access Interactive Docs

Open your browser to:

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

This is FastAPI's superpower!

You get a full interactive API documentation where you can:

See all endpoints
Read descriptions
Try requests directly in the browser
See request/response examples

Manual Testing Examples

Create a Task

curl -X POST "http://localhost:8000/tasks" \
  -H "Content-Type: application/json" \
  -d '{"title": "Learn FastAPI", "description": "Build a REST API"}'

Response:

{
  "id": "a3bb189e-8bf9-423b-bb46-98c7c5f3e5a7",
  "title": "Learn FastAPI",
  "description": "Build a REST API",
  "completed": false,
  "created_at": "2026-03-05T10:30:00.123456",
  "updated_at": "2026-03-05T10:30:00.123456"
}

Get All Tasks

curl http://localhost:8000/tasks

Update a Task

curl -X PUT "http://localhost:8000/tasks/{task_id}" \
  -H "Content-Type: application/json" \
  -d '{"title": "Updated Title"}'

Toggle Completion

curl -X PATCH "http://localhost:8000/tasks/{task_id}/complete"

Delete a Task

curl -X DELETE "http://localhost:8000/tasks/{task_id}"

Key Concepts Learned

1. REST API Design

CRUD Operations mapped to HTTP methods:

Create → POST /tasks
Read → GET /tasks, GET /tasks/{id}
Update → PUT /tasks/{id}
Delete → DELETE /tasks/{id}

Proper HTTP Status Codes:

200 OK: Successful GET, PUT, PATCH
201 Created: Successful POST
204 No Content: Successful DELETE
404 Not Found: Resource doesn't exist
422 Unprocessable Entity: Validation failed (automatic!)

2. Pydantic Validation

Pydantic handles all this automatically:

# Too long title (>100 chars)
{"title": "A" * 101}
# Returns: 422 with error message

# Missing required field
{"description": "Only description"}
# Returns: 422 "title is required"

# Wrong type
{"title": 123}
# Returns: 422 "title must be string"

3. Type Hints

Type hints make code:

Self-documenting: Clear what each variable should be
Editor-friendly: Better autocomplete
FastAPI-compatible: Framework uses them for validation

4. Async/Await

While we didn't use await in this project, using async def prepares our code for:

Database operations
External API calls
File operations
Concurrent request handling

Common Mistakes and How to Avoid Them

Mistake 1: Mutable Default Arguments

# ❌ WRONG
def create_task(tags: list = []):
    tags.append("new")  # Modifies the same list every time!

# ✅ CORRECT
def create_task(tags: Optional[List[str]] = None):
    if tags is None:
        tags = []
    tags.append("new")

Mistake 2: Not Checking Existence Before Update

# ❌ WRONG
@app.put("/tasks/{task_id}")
async def update_task(task_id: str, task_update: TaskUpdate):
    # What if task_id doesn't exist? KeyError!
    tasks_db[task_id]["title"] = task_update.title

# ✅ CORRECT (using our helper)
@app.put("/tasks/{task_id}")
async def update_task(task_id: str, task_update: TaskUpdate):
    task = get_task_or_404(task_id)  # Returns 404 if not found
    task["title"] = task_update.title

Mistake 3: Wrong HTTP Methods

# ❌ WRONG
@app.get("/tasks/{task_id}/delete")  # Should be DELETE, not GET

# ✅ CORRECT
@app.delete("/tasks/{task_id}")

Mistake 4: Not Using Status Codes

# ❌ WRONG
@app.post("/tasks")  # Returns 200 by default

# ✅ CORRECT
@app.post("/tasks", status_code=status.HTTP_201_CREATED)

Next Steps and Improvements

Immediate Enhancements

Add filtering:

   @app.get("/tasks")
   async def get_tasks(completed: Optional[bool] = None):
       if completed is not None:
           return [t for t in tasks_db.values() if t["completed"] == completed]
       return list(tasks_db.values())

Add pagination:

   @app.get("/tasks")
   async def get_tasks(skip: int = 0, limit: int = 10):
       tasks = list(tasks_db.values())
       return tasks[skip : skip + limit]

Add search:

   @app.get("/tasks")
   async def get_tasks(search: Optional[str] = None):
       if search:
           return [t for t in tasks_db.values() 
                   if search.lower() in t["title"].lower()]
       return list(tasks_db.values())

Future Learning Path

Day 2-5: Database integration (SQLAlchemy + PostgreSQL)
Day 6-10: Authentication and authorization (JWT, OAuth2)
Day 11-15: Testing (pytest, coverage)
Day 16-20: Docker deployment

Real-World Applications

This exact pattern is used for:

Todo apps (like we built)
Blog APIs (posts, comments)
E-commerce (products, orders)
Social media (posts, likes)
Project management (tasks, projects)

The CRUD pattern is fundamental to web development!

Conclusion

Today I learned:

✅ FastAPI basics and why it's gaining popularity
✅ How to build a complete REST API from scratch
✅ Pydantic models for automatic validation
✅ Proper HTTP methods and status codes
✅ Error handling best practices
✅ How to structure a professional API

Time Investment: 2.5 hours
Knowledge Gained: Production-ready API skills

Resources

Full Code

The complete code is available on my GitHub:
100-days-of-python/days/001

Tomorrow's Challenge: Adding PostgreSQL database integration and learning SQLAlchemy ORM.

Following my 100 Days journey? Connect with me on Twitter and LinkedIn!

Day 1/100 complete! ✅

Tags: #100DaysOfCode #Python #FastAPI #BackendDevelopment #WebDevelopment #RestAPI #LearningInPublic

No-code tools for SQL development

archi-jain — Fri, 27 Jan 2023 19:53:29 +0000

There are several tools that can make development with SQL easier without the need to write code:

SQL Management Studio: This is a graphical user interface (GUI) tool that allows developers to interact with databases, create and modify tables, run queries and manage data.

SQL Workbench: This is a free, open-source SQL client that provides a GUI for working with databases. It is available for Windows, Linux, and Mac.

DBVisualizer: This is a database management tool that allows developers to work with multiple databases simultaneously and includes features such as a SQL editor, data browsing, and visual data modeling.

Toad: This is a commercial tool that provides a GUI for working with databases, including a SQL editor, data browsing, and visual data modeling.

Navicat: This is a commercial SQL client that provides a GUI for working with databases, including a SQL editor, data browsing, and visual data modeling.

DataGrip: This is a commercial SQL client that provides a GUI for working with databases, including a SQL editor, data browsing, and visual data modeling.

All these tools are designed to provide an easy and intuitive way to work with SQL, allowing developers to quickly and efficiently manage databases, run queries and analyze data without the need to write code.

Getting started with Reactjs

archi-jain — Fri, 27 Jan 2023 18:13:12 +0000

Here are a few tips for someone who is just getting started with React

Learn the basics of JavaScript: React is a JavaScript library, so it's important to have a solid understanding of JavaScript before diving into React. Make sure you understand concepts like variables, functions, and objects.
Understand the component-based architecture: React is based on a component-based architecture, where the UI is divided into small, reusable components. Understanding this concept is essential for building efficient and maintainable React applications.
Start with the basics: React has a gentle learning curve, but it can be overwhelming at first. Start with the basics, such as creating and rendering components, handling events, and passing data between components.
Practice, practice, practice: The best way to learn React is by building projects with it. Try building small projects, such as a to-do list or a weather app, to get a feel for how React works and how to structure your code.

5. Learn JSX: JSX is a syntax extension for JavaScript that allows you to write HTML-like elements in your JavaScript code. It's used in React to describe the structure of a component's UI.

Understand the Virtual DOM: React uses a virtual DOM to improve the performance of updates and re-renders. Understanding how the virtual DOM works and how it interacts with the actual DOM can help you write more efficient React code.
Learn about state and props: React components have two types of data: props and state. Props are passed from a parent component to a child component, and state is managed within a component. Understanding the difference between props and state is important for building correct and efficient React applications.
Familiarize yourself with the React Developer Tools: React Developer Tools is a browser extension that allows you to inspect and debug your React code. It's a great tool for understanding how your components are structured and how they interact with each other.
Learn about React hooks: React Hooks are a way to use state and other React features in functional components. It's a powerful feature that can help you write more readable and reusable code.
Keep an eye on the community: React is a rapidly-evolving library, and new features and best practices are being introduced all the time. Keep an eye on the React community, by following blogs, forums, and social media, to stay up-to-date with the latest developments.

5 habits that can make you a good developer

archi-jain — Thu, 26 Jan 2023 12:12:52 +0000

Developing good habits is essential for a developer to be productive, efficient, and successful.
So, here are 5 must-have habits for a good developer

Continual learning: Technology is constantly evolving, and it's essential for developers to stay current with the latest tools, technologies, and best practices. This can involve taking online courses, reading technical books, and experimenting with new technologies.
Writing clean and readable code: Writing code that is easy to read and understand is essential for both the developer and other team members who may need to work on the code in the future. This includes using meaningful variable names, commenting code, and following a consistent coding style.
Time management: Developers often have a lot of tasks to complete, and it's essential to manage time effectively to meet deadlines and prioritize tasks. This can involve using tools such as calendars, to-do lists, and project management software, and learning how to estimate how long a task will take.
Focus and concentration: Developing the ability to focus and concentrate for long periods of time is essential for developers to be productive. This can involve setting up a comfortable work environment, minimizing distractions, and using techniques such as the Pomodoro Technique to increase focus and concentration.
Collaboration and teamwork: Developers often work in teams, and it's essential to have strong collaboration and teamwork skills to effectively communicate and work with other team members. This can involve using tools such as version control, issue tracking, and agile development methodologies to work with other team members.

7 Techniques for optimizing JavaScript performance and reducing load times

archi-jain — Tue, 24 Jan 2023 18:52:47 +0000

There are several techniques you can use to optimize the performance of your JavaScript code and reduce load times. Here are a few

Minification: This is the process of removing unnecessary characters from your code (such as whitespace, comments, and newlines) to reduce its file size. Minifying your code can help it load faster, especially on slow connections.
Bundling: This is the process of combining multiple JavaScript files into a single file. By reducing the number of requests the browser needs to make to load your code, bundling can help improve load times.
Lazy loading: This is a technique where you only load the JavaScript that is needed for the current view or user action. This can help improve the initial load time of your website, as well as the performance of the website as the user interacts with it.
Code splitting: Similar to lazy loading, it's a technique where you can split your code into multiple chunks and load them only when needed. This way, your initial load time will be faster, and users won't need to download the entire codebase to run your website.
Using a Content Delivery Network (CDN): CDN is a network of servers that are distributed around the world to deliver content to users based on their geographic location. By hosting your JavaScript on a CDN, you can reduce the distance that the data has to travel, and therefore, improve load times.
Using a JavaScript framework/library: Popular frameworks and libraries like React, Angular, and Vue.js can help you write better-structured and more maintainable code, as well as provide performance optimizations.
Properly Optimizing Images: Optimizing images is not directly related to JavaScript, but it's essential to mention it, as images can take a significant amount of time to load. Optimizing images by compressing or resizing them can greatly improve load times for your website.

Facts no one knows about programmers

archi-jain — Mon, 23 Jan 2023 21:24:14 +0000

Here are a few fun facts about coders or programmers that you may not know about:

Many early computer programmers were women: During the 1950s and 1960s, when computers were first being developed, many of the programmers were women. This is because computers were seen as more of a "girl's job," and many women were hired to work as "computers" (people who did calculations by hand) before computers were even invented.
The first computer programmer was a woman: Ada Lovelace, an English mathematician and writer, is considered to be the world's first computer programmer for her work on Charles Babbage's Analytical Engine in the 19th century. She wrote the first algorithm intended to be processed by the machine.
Some programmers use "rubber duck debugging": This is a technique where a programmer explains their code, line by line, to a rubber duck or any other inanimate object. The act of explaining the code helps the programmer to find the errors or bugs in their code.
The term "debugging" comes from an actual bug: In the early days of computing, bugs were not just a metaphorical term for errors in the code, but actual bugs that would crawl into the computer and cause problems. The term "debugging" comes from the practice of physically removing bugs from the computer.
The first computer virus was created by a 15-year-old: In 1986, a 15-year-old boy named Rich Skrenta created the first computer virus, called Elk Cloner. He wrote it as a joke and distributed it on floppy disks to his friends.
The first computer game was written by a programmer: The first computer game was written in 1958 by William Higinbotham, a physicist and programmer, called Tennis for Two. It was a simple game that used an oscilloscope as a display and was played on an analog computer.
The term "hacker" was originally a compliment: In the early days of computing, a "hacker" was someone who was skilled and clever with computers. The term has since taken on a negative connotation, but it originally was a compliment.