FastAPI is a modern, high-performance web framework for building APIs with Python. It provides automatic data validation, serialization, and interactive documentation, making it an excellent choice for developing reliable and maintainable web services.
In this tutorial, you'll build a task management API using FastAPI and SQLite as the database backend. We'll leverage SQLModel for database interactions and Pydantic for data validation, implementing CRUD operations: creating, reading, updating, and deleting tasks.
Prerequisites
Before getting started, ensure you have:
- Python installed on your system (preferably Python 3.13 or higher)
- A basic understanding of Python and web development concepts
Step 1 — Setting up the FastAPI project
In this section, you'll create the directory structure and install the necessary dependencies for the FastAPI project.
First, create a new directory for your project and navigate into it:
mkdir fastapi-task-api && cd fastapi-task-api
Next, create a virtual environment and activate it:
python3 -m venv venv
source venv/bin/activate
Install FastAPI and other dependencies:
pip install fastapi uvicorn sqlmodel python-dotenv
Here's a breakdown of these packages:
FastAPI
– The web framework for building APIs.Uvicorn
– An ASGI server required to run FastAPI applications.SQLModel
– A library that combines SQLAlchemy and Pydantic for database interactions.python-dotenv
– A package for working with environment variables.
Create a new file called main.py
and set up a basic FastAPI 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 script initializes a FastAPI application with metadata for the automatic documentation. It registers a simple route at /
, which responds with a JSON message.
When executed directly, the script runs the application using Uvicorn with the reload option enabled for development.
Now, run the application:
python main.py
You should see output similar to:
INFO: Will watch for changes in these directories: ['/Users/stanley/fast-api/fastapi-task-api']
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: Waiting for application startup.
INFO: Application startup complete.
This means your FastAPI server is running. Open http://127.0.0.1:8000/
in a browser to see the welcome message, or use curl
to test the API:
curl http://127.0.0.1:8000/
The output will be:
{
"message": "Welcome to the Task Management API"
}
FastAPI also provides automatic interactive documentation. Visit http://127.0.0.1:8000/docs
to explore the Swagger UI documentation for your API:
At this point, your FastAPI application is set up and running. In the next step, you'll configure the database using SQLModel.
Step 2 — Configuring the database with SQLModel
With the FastAPI application set up, the next step is to integrate a database to store and manage tasks. SQLModel simplifies database interactions by combining SQLAlchemy's ORM capabilities with Pydantic's validation features.
First, let's organize our project structure. Create the following directories:
mkdir -p app app/models app/database
This command creates the necessary directories for the project. The -p
flag ensures parent directories are created as needed and prevents errors if directories already exist.
Following that, create the following files in the newly created directories:
touch app/__init__.py app/models/__init__.py app/database/__init__.py
This command creates empty Python files that mark each directory as a package, enabling clean imports between modules and establishing the application's namespace hierarchy.
These files can later contain initialization code, shared utilities, or package metadata.
Next, create a file for database configuration:
touch app/database/config.py
Now, update the app/database/config.py
file with the following code:
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 the database connection using SQLModel. It defines two key functions:
create_db_and_tables()
: Creates all the tables defined in our models.get_session()
: A dependency function that provides a database session for our endpoints.
The echo=True
parameter enables SQL query logging for development, and connect_args={"check_same_thread": False}
allows SQLite to work with FastAPI's asynchronous nature.
Now, let's create a task model using SQLModel. Create a new file:
touch app/models/task.py
Update the app/models/task.py
file with the following code:
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
The TaskBase
class serves as the foundation, containing shared fields like title
, description
, priority
, and completed
status. The Task
class extends TaskBase
and represents the database model, adding a unique id
field generated using UUID, along with created_at
and updated_at
timestamps.
Three additional models are defined to handle different operations. The TaskCreate
class validates data when creating new tasks, while the TaskRead
class structures the data when retrieving tasks from the database.
Lastly, the TaskUpdate
class allows partial updates, where fields can be optionally modified.
Using inheritance and type hinting enables automatic validation, serialization, and documentation.
Following that, update the main.py
file to include the database initialization:
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)
You've added an event handler using @app.on_event("startup")
that calls create_db_and_tables()
when the application starts, ensuring that all necessary tables exist.
When you save the file, the application server should restart automatically. If it doesn’t, you can manually restart it by running the following command:
python main.py
You'll see additional SQL output in the console showing that the tables are being created:
INFO: Started server process [7654]
INFO: Waiting for application startup.
2025-02-28 14:15:37.123 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2025-02-28 14:15:37.124 INFO sqlalchemy.engine.Engine PRAGMA main.table_info("task")
2025-02-28 14:15:37.125 INFO sqlalchemy.engine.Engine [raw sql] ()
2025-02-28 14:15:37.126 INFO sqlalchemy.engine.Engine CREATE TABLE task (
id VARCHAR NOT NULL,
title VARCHAR NOT NULL,
description VARCHAR,
priority INTEGER NOT NULL,
completed BOOLEAN NOT NULL,
created_at TIMESTAMP NOT NULL,
updated_at TIMESTAMP NOT NULL,
PRIMARY KEY (id)
)
2025-02-28 14:15:37.127 INFO sqlalchemy.engine.Engine [no key 0.00031s] ()
2025-02-28 14:15:37.128 INFO sqlalchemy.engine.Engine CREATE INDEX ix_task_id ON task (id)
...
INFO: Application startup complete.
The database is now integrated with your FastAPI application. You will see a tasks.db
file in your project directory.
...
├── app
│ ├── ...
├── main.py
├── tasks.db
Now that you have set up the database, it's time to build the functionality that interacts with it. In the next step, you'll implement the endpoint for creating tasks.
Step 3 — Implementing the POST endpoint for creating tasks
Now that your models and database configuration are set up, let's implement the first API endpoint for creating new tasks. You'll organize your code by creating a separate module for all task-related routes.
To organize your route handlers, first create the necessary directory and files with the following command:
mkdir app/routes && touch app/routes/__init__.py app/routes/tasks.py
Now, update the app/routes/tasks.py
file with the following code:
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 defines a POST
endpoint for creating tasks. When a request is sent, FastAPI validates the input using the TaskCreate
model.
The function then converts the request data into a database-compatible Task
model and saves it to the database using SQLModel.
Once committed, the newly created task is returned as a response. This structured approach ensures efficient task management while maintaining data integrity.
Next, update the main.py
file to include the task routes:
from fastapi import FastAPI
from app.database.config import create_db_and_tables
from app.routes.tasks import router as tasks_router
app = FastAPI(
...
)
# Include routers
app.include_router(tasks_router)
@app.get("/")
async def root():
...
With the task routes included, save your file and use curl
to create a new task.
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 receive 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"
}
You can also test the validation by trying to create a task with an invalid priority (outside the 1-5 range):
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 receive a validation error:
{
"detail": [
{
"type": "less_than_equal",
"loc": [
"body",
"priority"
],
"msg": "Input should be less than or equal to 5",
"input": 10,
"ctx": {
"le": 5
}
}
]
}
This shows that Pydantic's validation is working automatically to ensure data consistency.
Now, visit http://127.0.0.1:8000/docs
in your browser to see the interactive Swagger UI, which includes information about your task creation endpoint.
Now that you can create tasks, it's time to add functionality to retrieve them.
Step 4 — Implementing GET endpoints for retrieving tasks
Now that you can create tasks, let's implement endpoints to retrieve them. We'll start with an endpoint for listing all tasks with optional filtering.
Update the app/routes/tasks.py
file to add the first GET endpoint:
...
@router.post("/", response_model=TaskRead, status_code=status.HTTP_201_CREATED)
def create_task(*, session: Session = Depends(get_session), task: TaskCreate):
...
@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
The highlighted code defines a GET endpoint that retrieves a list of tasks with optional filtering and pagination. It is registered under /api/tasks/
and returns a list of tasks formatted using the TaskRead
model.
To enable pagination, the offset
and limit
parameters allow clients to skip a specified number of tasks and limit the number of results returned.
Additionally, the completed
parameter enables filtering tasks by their completion status. The query is adjusted to return only completed or incomplete tasks if provided.
The endpoint constructs a database query using select(Task)
, applies filters if necessary, and retrieves the results using session.exec()
.
The final list of tasks is then returned, ensuring efficient task retrieval with flexibility for filtering and pagination.
Now, test the endpoint for retrieving all tasks. First, let's create a few tasks to have some data to work with:
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
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
Next, test retrieving all tasks:
curl http://127.0.0.1:8000/api/tasks/ | python3 -m json.tool
You should see a response with all your tasks:
[
{
...
},
{
"title": "Learn FastAPI",
"description": "Complete the FastAPI tutorial and build a project",
"priority": 2,
"completed": false,
"id": "1ff677f3-82e4-47a3-a228-58650136622b",
"created_at": "2025-02-28T10:58:15.735311",
"updated_at": "2025-02-28T10:58:15.735325"
},
{
"title": "Write Documentation",
"description": "Document the API endpoints and models",
"priority": 3,
"completed": true,
"id": "d5477b21-1b1a-4745-b2b5-01da0f65a679",
"created_at": "2025-02-28T10:59:22.476886",
"updated_at": "2025-02-28T10:59:22.476901"
}
]
You can also test filtering for completed tasks:
curl "http://127.0.0.1:8000/api/tasks/?completed=true" | python3 -m json.tool
This should return only the tasks marked as completed:
[
{
"title": "Write Documentation",
"description": "Document the API endpoints and models",
"priority": 3,
"completed": true,
"id": "d5477b21-1b1a-4745-b2b5-01da0f65a679",
"created_at": "2025-02-28T10:59:22.476886",
"updated_at": "2025-02-28T10:59:22.476901"
}
]
Now that you have confirmed that the list endpoint works, you will implement the endpoint to retrieve a specific task by ID.
Add the following code to 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
Test retrieving a specific task by its ID. Make sure to use an ID from your previous responses:
curl "http://127.0.0.1:8000/api/tasks/d5477b21-1b1a-4745-b2b5-01da0f65a679" | python3 -m json.tool
You should receive the specific task:
{
"title": "Write Documentation",
"description": "Document the API endpoints and models",
"priority": 3,
"completed": true,
"id": "d5477b21-1b1a-4745-b2b5-01da0f65a679",
"created_at": "2025-02-28T10:59:22.476886",
"updated_at": "2025-02-28T10:59:22.476901"
}
Try requesting a non-existent task to see the 404 error handling:
curl "http://127.0.0.1:8000/api/tasks/non-existent-id" | python3 -m json.tool
{
"detail": "Task with ID non-existent-id not found"
}
The interactive documentation at http://127.0.0.1:8000/docs
is also automatically updated to include these new endpoints, complete with all the parameters and response models:
Now that you have implemented task retrieval, the next step is to provide users with the ability to update existing tasks efficiently.
Step 5 — Implementing the PUT endpoint for updating tasks
Now that you can create and retrieve tasks, the next step is to allow updates. You'll implement a PUT endpoint for performing full updates, meaning that all task fields must be provided in the request body. This ensures that the entire task is replaced with the new data, maintaining consistency across updates.
Update the app/routes/tasks.py
file to add the PUT endpoint:
...
@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
Test the PUT endpoint for full updates. This requires providing all task fields, even if they remain unchanged(Remember to replace with your actual task ID):
curl -X PUT "http://127.0.0.1:8000/api/tasks/<your_task_id>" \
-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
You should see a response with all fields updated:
{
"title": "Learn FastAPI in Depth",
"description": "Complete the advanced FastAPI tutorial",
"priority": 4,
"completed": true,
"id": "d5477b21-1b1a-4745-b2b5-01da0f65a679",
"created_at": "2025-02-28T10:59:22.476886",
"updated_at": "2025-02-28T11:14:41.070992"
}
Notice that the updated_at
timestamp has been automatically updated to reflect the modification time.
Now that you've confirmed the PUT endpoint works for full updates, check the interactive documentation at http://127.0.0.1:8000/docs
:
The documentation now includes the update endpoint, providing a clear interface for testing and verifying task updates directly from the browser.
Step 6 — Implementing the DELETE endpoint for removing tasks
To complete the CRUD functionality of your task API, the final step is to implement a DELETE endpoint. This will allow users to remove tasks from the database.
Update the app/routes/tasks.py
file to add the delete endpoint:
...
@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 DELETE
endpoint allows users to remove tasks by providing a valid task ID. When called, it first retrieves the task from the database. If the task doesn't exist, it raises a 404 Not Found
error. Otherwise, it deletes the task and commits the changes. The endpoint returns a 204 No Content
status code, indicating that the operation was successful without requiring a response body.
Now, test this endpoint by deleting one of your tasks(replace with your actual task ID):
curl -X DELETE "http://127.0.0.1:8000/api/tasks/<your_task_id>" -v
The -v
(verbose) flag will show the response headers, including the status code:
* Trying 127.0.0.1:8000...
* Connected to 127.0.0.1 (127.0.0.1) port 8000
> DELETE /api/tasks/d5477b21-1b1a-4745-b2b5-01da0f65a679 HTTP/1.1
> Host: 127.0.0.1:8000
> User-Agent: curl/8.7.1
> Accept: */*
>
* Request completely sent off
< HTTP/1.1 204 No Content
< date: Fri, 28 Feb 2025 11:23:51 GMT
< server: uvicorn
< content-type: application/json
<
* Connection #0 to host 127.0.0.1 left intact
The HTTP status code 204 confirms that the task was successfully deleted. You can verify this by trying to retrieve the deleted task:
curl http://127.0.0.1:8000/api/tasks/<your_task_id>
You should receive a 404 Not Found response:
{"detail":"Task with ID d5477b21-1b1a-4745-b2b5-01da0f65a679 not found"}
With the DELETE
endpoint in place, your task management API now fully supports Create, Read, Update, and Delete (CRUD) operations.
Final thoughts
This article has guided you through building a production-ready RESTful API using FastAPI while ensuring type safety and data validation throughout your application.
To deepen your understanding of FastAPI and enhance your skills, consider exploring the official documentation for FastAPI, SQLModel, and Pydantic. These resources provide valuable insights, best practices, and advanced techniques to help you build even more robust and scalable applications.
Thanks for reading and happy coding!
Make your mark
Join the writer's program
Are you a developer and love writing and sharing your knowledge with the world? Join our guest writing program and get paid for writing amazing technical guides. We'll get them to the right readers that will appreciate them.
Write for us
Build on top of Better Stack
Write a script, app or project on top of Better Stack and share it with the world. Make a public repository and share it with us at our email.
community@betterstack.comor submit a pull request and help us build better products for everyone.
See the full list of amazing projects on github