FastAPI is a modern Python web framework that helps you build APIs quickly and easily. Think of an API as a bridge that allows different applications to talk to each other. FastAPI makes this process simple by automatically handling data validation and creating beautiful documentation for your API.
Today we'll build a task management API from scratch. This is like creating a digital to-do list where you can add new tasks, view existing ones, update them, and delete completed tasks. We'll use SQLite as our database because it's lightweight and perfect for beginners.
Prerequisites
Before we start coding, make sure you have:
Python installed on your computer (version 3.13 or newer works best)
Some basic knowledge of Python programming
Understanding of what APIs are and how they work
Don't worry if you're new to APIs - we'll explain everything step by step.
Step 1: Setting up your project
Let's start by creating a new folder for our project. Open your terminal or command prompt and run these commands:
mkdir fastapi-task-api
cd fastapi-task-api
Now we need to create a virtual environment. Think of this as a separate space where we can install our project's dependencies without affecting other Python projects on your computer:
python3 -m venv venv
source venv/bin/activate
After activating the virtual environment, let's install the packages we need:
pip install fastapi uvicorn sqlmodel python-dotenv
Here's what each package does:
FastAPI: The main framework for building our API
Uvicorn: The server that will run our application
SQLModel: Makes working with databases much easier
python-dotenv: Helps manage configuration settings
Now let's create our first file called main.py
. This will be the entry point of our application:
from fastapi import FastAPI
app = FastAPI(
title="Task Management API",
description="API for managing tasks with FastAPI, SQLModel, and Pydantic",
version="0.1.0"
)
@app.get("/")
async def root():
"""Health check endpoint for the API."""
return {"message": "Welcome to the Task Management API"}
if __name__ == "__main__":
import uvicorn
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)
This code creates a basic FastAPI application. The @app.get("/")
decorator creates a route that responds to GET requests at the root URL. When someone visits our API, they'll see a welcome message.
Let's test our application by running it:
python main.py
You should see output like this:
INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
INFO: Started reloader process [21967] using StatReload
INFO: Started server process [21969]
INFO: Application startup complete.
Great! Open your web browser and go to http://127.0.0.1:8000/
. You should see a JSON response with your welcome message.
Here's something cool: FastAPI automatically creates interactive documentation for your API. Visit http://127.0.0.1:8000/docs
to see the Swagger UI documentation. This is where you can test your API endpoints directly from the browser.
Step 2: Setting up the database
Now we need to add a database to store our tasks. We'll use SQLModel, which makes database operations much simpler than traditional methods.
First, let's organize our project by creating some folders. Run these commands:
mkdir -p app app/models app/database
touch app/__init__.py app/models/__init__.py app/database/__init__.py
These commands create folders and empty Python files. The __init__.py
files tell Python that these folders are packages that can be imported.
Now let's create our database configuration. Create a file called app/database/config.py
:
import os
from sqlmodel import SQLModel, create_engine, Session
# Get database URL from environment variable or use default SQLite URL
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./tasks.db")
# Create SQLAlchemy engine
engine = create_engine(
DATABASE_URL,
echo=True, # Set to False in production
connect_args={"check_same_thread": False} # Only needed for SQLite
)
def create_db_and_tables():
"""Create all tables in the database."""
SQLModel.metadata.create_all(engine)
def get_session():
"""Create a new database session."""
with Session(engine) as session:
yield session
This code sets up our database connection. The create_db_and_tables()
function will create our database tables, and get_session()
provides a way for our API to connect to the database.
Next, let's create our task model. This defines what a task looks like in our database. Create a file called app/models/task.py
:
from datetime import datetime
from typing import Optional
from sqlmodel import Field, SQLModel
import uuid
def generate_uuid():
"""Generate a unique UUID for a task."""
return str(uuid.uuid4())
class TaskBase(SQLModel):
"""Base model for task data."""
title: str = Field(index=True)
description: Optional[str] = Field(default=None)
priority: int = Field(default=1, ge=1, le=5)
completed: bool = Field(default=False)
class Task(TaskBase, table=True):
"""Database model for tasks."""
id: str = Field(
default_factory=generate_uuid,
primary_key=True,
index=True
)
created_at: datetime = Field(default_factory=datetime.utcnow)
updated_at: datetime = Field(
default_factory=datetime.utcnow,
sa_column_kwargs={"onupdate": datetime.utcnow}
)
class TaskCreate(TaskBase):
"""Model for creating a new task."""
pass
class TaskRead(TaskBase):
"""Model for reading a task."""
id: str
created_at: datetime
updated_at: datetime
class TaskUpdate(SQLModel):
"""Model for updating a task."""
title: Optional[str] = None
description: Optional[str] = None
priority: Optional[int] = None
completed: Optional[bool] = None
Let me explain what each class does:
TaskBase: Contains the common fields that all tasks have (title, description, priority, completed status)
Task: The main database model with additional fields like ID and timestamps
TaskCreate: Used when creating new tasks to validate input data
TaskRead: Used when returning task data to users
TaskUpdate: Used when updating existing tasks, allows partial updates
Now let's update our main.py
file to initialize the database when the application starts:
from fastapi import FastAPI
from app.database.config import create_db_and_tables
app = FastAPI(
title="Task Management API",
description="API for managing tasks with FastAPI, SQLModel, and Pydantic",
version="0.1.0"
)
@app.get("/")
async def root():
"""Health check endpoint for the API."""
return {"message": "Welcome to the Task Management API"}
@app.on_event("startup")
def on_startup():
"""Initialize database when the application starts."""
create_db_and_tables()
if __name__ == "__main__":
import uvicorn
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)
The @app.on_event("startup")
decorator tells FastAPI to run the create_db_and_tables()
function when the application starts. This ensures our database tables are created automatically.
When you save the file, your server should restart automatically. You'll see SQL output in the console showing that the database tables are being created. You'll also notice a new file called tasks.db
in your project directory - this is your SQLite database file.
Step 3: Creating tasks
Now let's add the ability to create new tasks. We'll organize our API endpoints in a separate file to keep things clean and manageable.
Create the routes directory and files:
mkdir app/routes
touch app/routes/__init__.py app/routes/tasks.py
Now let's add our first API endpoint in app/routes/tasks.py
:
from fastapi import APIRouter, Depends, HTTPException, status
from sqlmodel import Session, select
from typing import List
from app.database.config import get_session
from app.models.task import Task, TaskCreate, TaskRead
router = APIRouter(prefix="/api/tasks", tags=["Tasks"])
@router.post("/", response_model=TaskRead, status_code=status.HTTP_201_CREATED)
def create_task(*, session: Session = Depends(get_session), task: TaskCreate):
"""
Create a new task.
- **title**: Required. The title of the task.
- **description**: Optional. Detailed description of the task.
- **priority**: Optional. Priority level (1-5), defaults to 1.
- **completed**: Optional. Task completion status, defaults to False.
"""
# Convert TaskCreate model to Task model
db_task = Task.from_orm(task)
# Add to database
session.add(db_task)
session.commit()
session.refresh(db_task)
return db_task
This code creates a POST endpoint that accepts task data. Here's what happens step by step:
FastAPI automatically validates the incoming data using our
TaskCreate
modelWe convert the validated data to a
Task
object that can be saved to the databaseWe add the task to the database session and commit the changes
We return the created task, which FastAPI automatically converts to JSON
Now we need to tell our main application about these routes. Update your main.py
file:
from fastapi import FastAPI
from app.database.config import create_db_and_tables
from app.routes.tasks import router as tasks_router
app = FastAPI(
title="Task Management API",
description="API for managing tasks with FastAPI, SQLModel, and Pydantic",
version="0.1.0"
)
# Include routers
app.include_router(tasks_router)
@app.get("/")
async def root():
"""Health check endpoint for the API."""
return {"message": "Welcome to the Task Management API"}
@app.on_event("startup")
def on_startup():
"""Initialize database when the application starts."""
create_db_and_tables()
if __name__ == "__main__":
import uvicorn
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)
Now let's test our task creation endpoint. You can use curl from your terminal:
curl -X POST http://127.0.0.1:8000/api/tasks/ \
-H "Content-Type: application/json" \
-d '{"title":"Learn FastAPI","description":"Complete the FastAPI tutorial and build a project","priority":2}' \
| python3 -m json.tool
You should get a response like this:
{
"title": "Learn FastAPI",
"description": "Complete the FastAPI tutorial and build a project",
"priority": 2,
"completed": false,
"id": "5b499058-a5d6-4c85-8a88-0d1dd4600fac",
"created_at": "2025-02-28T10:44:55.434117",
"updated_at": "2025-02-28T10:44:55.434124"
}
Notice that FastAPI automatically generated a unique ID and timestamps for our task. Try creating a task with invalid data to see the validation in action:
curl -X POST http://127.0.0.1:8000/api/tasks/ \
-H "Content-Type: application/json" \
-d '{"title":"Invalid Task","priority":10}' \
| python3 -m json.tool
You'll get a validation error because priority must be between 1 and 5. This shows that our data validation is working correctly.
Step 4: Getting tasks
Now let's add endpoints to retrieve our tasks. We'll create two endpoints: one to get all tasks and another to get a specific task by its ID.
Add this code to your app/routes/tasks.py
file:
@router.get("/", response_model=List[TaskRead])
def read_tasks(
*,
session: Session = Depends(get_session),
offset: int = 0,
limit: int = 100,
completed: bool = None
):
"""
Retrieve a list of tasks with optional filtering.
- **offset**: Number of tasks to skip (for pagination).
- **limit**: Maximum number of tasks to return (for pagination).
- **completed**: Filter by completion status.
"""
query = select(Task)
# Apply completion status filter if provided
if completed is not None:
query = query.where(Task.completed == completed)
# Apply pagination
tasks = session.exec(query.offset(offset).limit(limit)).all()
return tasks
This endpoint allows you to:
Get all tasks by default
Filter tasks by completion status (completed=true or completed=false)
Use pagination to limit results (useful when you have many tasks)
Let's create another task first so we have some data to work with:
curl -X POST http://127.0.0.1:8000/api/tasks/ \
-H "Content-Type: application/json" \
-d '{"title":"Write Documentation","description":"Document the API endpoints and models","priority":3,"completed":true}' \
| python3 -m json.tool
Now test retrieving all tasks:
curl http://127.0.0.1:8000/api/tasks/ | python3 -m json.tool
You should see a list of all your tasks. You can also filter for completed tasks:
curl "http://127.0.0.1:8000/api/tasks/?completed=true" | python3 -m json.tool
Now let's add an endpoint to get a specific task by its ID. Add this to your app/routes/tasks.py
:
@router.get("/{task_id}", response_model=TaskRead)
def read_task(*, session: Session = Depends(get_session), task_id: str):
"""
Retrieve a specific task by ID.
- **task_id**: The unique identifier of the task.
"""
task = session.get(Task, task_id)
if not task:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Task with ID {task_id} not found"
)
return task
This endpoint takes a task ID from the URL and returns that specific task. If the task doesn't exist, it returns a 404 error.
Test it by getting a specific task (replace the ID with one from your previous responses):
curl "http://127.0.0.1:8000/api/tasks/your-task-id-here" | python3 -m json.tool
Try requesting a task that doesn't exist to see the error handling:
curl "http://127.0.0.1:8000/api/tasks/non-existent-id" | python3 -m json.tool
Step 5: Updating tasks
Now let's add the ability to update existing tasks. We'll use a PUT endpoint, which means you need to provide all the task fields when updating.
Add this code to your app/routes/tasks.py
:
@router.put("/{task_id}", response_model=TaskRead)
def update_task(
*,
session: Session = Depends(get_session),
task_id: str,
task: TaskCreate
):
"""
Update a task completely.
- **task_id**: The unique identifier of the task.
- Request body: All task fields (even unchanged ones).
"""
db_task = session.get(Task, task_id)
if not db_task:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Task with ID {task_id} not found"
)
# Update all attributes
task_data = task.dict()
for key, value in task_data.items():
setattr(db_task, key, value)
# Update in database
session.add(db_task)
session.commit()
session.refresh(db_task)
return db_task
This endpoint works by:
Finding the task by ID
Updating all the task's fields with the new data
Saving the changes to the database
Returning the updated task
Test the update endpoint (replace the ID with one of your actual task IDs):
curl -X PUT "http://127.0.0.1:8000/api/tasks/your-task-id-here" \
-H "Content-Type: application/json" \
-d '{"title":"Learn FastAPI in Depth","description":"Complete the advanced FastAPI tutorial","priority":4,"completed":true}' \
| python3 -m json.tool
Notice that the updated_at
timestamp automatically changes when you update a task. This happens because we configured it in our Task model.
Step 6: Deleting tasks
Finally, let's add the ability to delete tasks. This completes our CRUD operations (Create, Read, Update, Delete).
Add this code to your app/routes/tasks.py
:
@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_task(*, session: Session = Depends(get_session), task_id: str):
"""
Delete a task.
- **task_id**: The unique identifier of the task.
"""
task = session.get(Task, task_id)
if not task:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Task with ID {task_id} not found"
)
# Delete from database
session.delete(task)
session.commit()
# Return no content
return None
This endpoint finds the task by ID and deletes it from the database. It returns a 204 No Content status code, which means the operation was successful but there's no content to return.
Test the delete endpoint:
curl -X DELETE "http://127.0.0.1:8000/api/tasks/your-task-id-here" -v
The -v
flag shows you the response headers, including the 204 status code. To confirm the task was deleted, try to get it again:
curl http://127.0.0.1:8000/api/tasks/your-task-id-here
You should receive a 404 Not Found error, confirming that the task was successfully deleted.
Conclusion
You've successfully built a complete task management API using FastAPI! Your API now supports all the essential CRUD operations and includes automatic data validation, error handling, and interactive documentation. FastAPI's powerful features like automatic JSON serialization, type checking, and built-in documentation make it an excellent choice for building modern web APIs. You can now expand this foundation by adding features like user authentication, more complex data relationships, or deploying your API to the cloud.