FastAPI SQLModel CRUD: A Practical Example

Hey guys, let’s dive into the awesome world of building web APIs with Python! Today, we’re tackling a super common task: creating a FastAPI SQLModel CRUD example . If you’re not familiar, CRUD stands for Create, Read, Update, and Delete – the four fundamental operations you’ll perform on data in most applications. And when you combine FastAPI, a modern, fast (hence the name!), web framework for Python, with SQLModel, a library that elegantly merges Pydantic and SQLAlchemy, you get a seriously powerful and developer-friendly way to build your database-backed APIs. This isn’t just about theory; we’re going to walk through a practical, hands-on example so you can see exactly how it all comes together. We’ll be setting up a simple project, defining our data models, and then implementing each CRUD operation. So grab your favorite IDE, get ready to type some code, and let’s build something cool!

Setting Up Your Environment

Before we can start coding our FastAPI SQLModel CRUD example , we need to get our development environment set up. First things first, make sure you have Python installed. If not, head over to python.org and grab the latest version. Once Python is sorted, we’ll create a virtual environment. This is crucial for managing project dependencies and avoiding conflicts between different projects. Open your terminal or command prompt, navigate to where you want to create your project, and run the following commands:

python -m venv venv
# On Windows
venv\Scripts\activate
# On macOS/Linux
source venv/bin/activate

Now that our virtual environment is active, it’s time to install the necessary libraries. We’ll need fastapi for our web framework, uvicorn to run our FastAPI application, sqlalchemy for the database ORM magic, and sqlmodel to tie it all together. Let’s install them:

pip install fastapi uvicorn sqlalchemy sqlmodel

We also need a database. For this example, we’ll keep things simple and use SQLite, which is a file-based database and doesn’t require a separate server. If you plan to use PostgreSQL or MySQL for production, the concepts remain largely the same, but the database connection string will differ. To work with SQLite, we’ll also need aiosqlite if you intend to use async operations, which is highly recommended with FastAPI.

pip install aiosqlite

With all our dependencies installed, we’re ready to move on to defining our data models. This is where SQLModel shines, allowing us to define our database tables and our API request/response models in one go. This DRY (Don’t Repeat Yourself) principle is a huge win for developer productivity!

Defining Your Data Models with SQLModel

Now for the fun part: defining our data models! SQLModel makes this incredibly straightforward by letting us define a single Python class that serves as both our SQLAlchemy model (for database interactions) and our Pydantic model (for data validation and serialization in our API). This is a game-changer, honestly. For our FastAPI SQLModel CRUD example , let’s imagine we’re building a simple task management API. We’ll need a Todo item, which should have an ID, a title, a description, and a status (like ‘pending’ or ‘completed’).

Let’s create a file named models.py and define our Todo model. We’ll inherit from SQLModel and also specify table=True so SQLModel knows this class represents a database table. We’ll use type hints extensively, which is a cornerstone of modern Python and how SQLModel leverages Pydantic for validation.

from typing import Optional
from sqlmodel import Field, SQLModel

class TodoBase(SQLModel):
    title: str
    description: Optional[str] = None
    completed: bool = False

class Todo(TodoBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)

class TodoCreate(TodoBase):
    pass

class TodoRead(TodoBase):
    id: int

class TodoUpdate(TodoBase):
    title: Optional[str] = None
    description: Optional[str] = None
    completed: Optional[bool] = None

In this code snippet, you can see a few things:

• TodoBase : This class contains the common fields for our Todo item. It inherits from SQLModel and uses standard Python type hints. Optional[str] means the field can be a string or None , and bool = False sets a default value.
• Todo(TodoBase, table=True) : This is our main database model. By inheriting from TodoBase and setting table=True , we tell SQLModel to create a database table for this model. id: Optional[int] = Field(default=None, primary_key=True) defines our primary key, which will be auto-generated by the database.
• TodoCreate(TodoBase) : This model is used when a client creates a new Todo item. It inherits all fields from TodoBase . We don’t need an id here because it’s generated by the database.
• TodoRead(TodoBase) : This model is used when we send a Todo item back to the client. It includes the id that was generated.
• TodoUpdate(TodoBase) : This model is used when a client wants to update an existing Todo item. We make all fields optional here because a client might only want to update one field, not all of them.

This approach using SQLModel is super clean. You define your structure once, and it handles validation for API requests/responses and also maps directly to your database schema. Pretty neat, huh?

Setting Up the Database and Session

Alright, we’ve got our models defined. Now, let’s get our database set up and create a session to interact with it. For our FastAPI SQLModel CRUD example , we’ll stick with SQLite. We need a place to store our database file and a way to create database tables based on our Todo model.

Let’s create a file named database.py . This file will handle our database connection and session management. We’ll use SQLAlchemy’s create_engine to set up our connection and SQLModel.metadata.create_all to create the tables.

from sqlmodel import SQLModel, create_engine

DATABASE_FILE = "database.db"
DATABASE_CONNECTION_STRING = f"sqlite:///{DATABASE_FILE}"

engine = create_engine(DATABASE_CONNECTION_STRING, echo=True) # echo=True will log SQL queries

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

In this database.py file:

• We define the DATABASE_FILE and the DATABASE_CONNECTION_STRING for our SQLite database. The sqlite:/// prefix indicates a relative path to the database file.
• create_engine is used to establish the connection to our database. The echo=True argument is super handy during development because it prints all the SQL statements that SQLAlchemy executes, which helps a ton with debugging.
• The create_db_and_tables() function uses SQLModel.metadata.create_all(engine) to automatically create all the tables defined in our SQLModel classes (like our Todo table) when the application starts. This is a declarative way to manage your database schema.

Now, we need a way to get a database session within our API endpoints. FastAPI provides a dependency injection system that’s perfect for this. Let’s create a function that yields a database session and ensures it’s closed properly after the request is done.

Create a file named dependencies.py :

from typing import Generator
from sqlmodel import Session, create_engine

from database import engine

def get_session() -> Generator[Session, None, None]:
    with Session(engine) as session:
        yield session

This get_session function is a generator. When it’s called by FastAPI’s dependency injection, it creates a new SQLAlchemy Session using our engine . The with statement ensures that the session is automatically closed (committed or rolled back) when the request handling is finished, preventing resource leaks. This is a standard and robust way to manage database sessions in FastAPI applications.

Implementing the Create Operation

Let’s get started with the first part of our FastAPI SQLModel CRUD example : the Create operation. This is how we’ll add new Todo items to our database. We’ll create a main file, let’s call it main.py , where our FastAPI app will live.

from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import Session, select
from typing import List

from models import Todo, TodoCreate, TodoRead, TodoUpdate
from database import engine, create_db_and_tables
from dependencies import get_session

# Create database tables on startup
create_db_and_tables()

app = FastAPI()

@app.post("/todos/", response_model=TodoRead, status_code=201)
def create_todo(
    todo: TodoCreate,
    session: Session = Depends(get_session)
):
    db_todo = Todo.model_validate(todo)
    session.add(db_todo)
    session.commit()
    session.refresh(db_todo)
    return db_todo

Let’s break down what’s happening here:

• We import necessary modules: FastAPI , Depends , HTTPException from fastapi ; Session , select from sqlmodel ; List for type hinting; and our custom models, database setup, and dependencies.
• create_db_and_tables() is called immediately to ensure our database and tables exist when the app starts.
• app = FastAPI() initializes our FastAPI application.
• The @app.post("/todos/", response_model=TodoRead, status_code=201) decorator defines our API endpoint:
  • POST /todos/ : This endpoint will handle creating new todos.
  • response_model=TodoRead : Specifies that the response will conform to the TodoRead model, which includes the id .
  • status_code=201 : Sets the HTTP status code to 201 Created for successful resource creation.
• def create_todo(todo: TodoCreate, session: Session = Depends(get_session)) : This is our path operation function.
  • todo: TodoCreate : FastAPI automatically parses the incoming JSON request body and validates it against our TodoCreate Pydantic model.
  • session: Session = Depends(get_session) : This injects a database session into our function, thanks to our get_session dependency.
• db_todo = Todo.model_validate(todo) : We convert the incoming TodoCreate data into our Todo model, which is suitable for database insertion.
• session.add(db_todo) : Adds the new Todo object to the session.
• session.commit() : Commits the transaction to the database.
• session.refresh(db_todo) : Refreshes the db_todo object from the database. This is important to get any database-generated values, like the id .
• return db_todo : Returns the created Todo object (as a TodoRead model because of response_model ).

To test this, save the files ( models.py , database.py , dependencies.py , main.py ) in the same directory, and run uvicorn main:app --reload in your terminal. Then, you can use tools like curl or Postman to send a POST request to http://127.0.0.1:8000/todos/ with a JSON body like {"title": "Learn SQLModel", "description": "Build a CRUD app!"} .

Implementing the Read Operations

Next up in our FastAPI SQLModel CRUD example are the Read operations. We need ways to fetch one or all of our Todo items. We’ll add these to our main.py file.

First, let’s implement the endpoint to get all todos:

@app.get("/todos/", response_model=List[TodoRead])
def read_todos(session: Session = Depends(get_session)):
    statement = select(Todo)
    results = session.exec(statement).all()
    return results

And here’s the endpoint to get a single todo by its ID:

@app.get("/todos/{todo_id}", response_model=TodoRead)
def read_todo(todo_id: int, session: Session = Depends(get_session)):
    todo = session.get(Todo, todo_id)
    if todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")
    return todo

Let’s break these down:

• read_todos endpoint ( @app.get("/todos/", response_model=List[TodoRead]) ) :

  • This endpoint uses the GET HTTP method and maps to the /todos/ URL.
  • response_model=List[TodoRead] indicates that the response will be a list of TodoRead objects.
  • Inside the function, statement = select(Todo) creates a SQLAlchemy SELECT statement to fetch all records from the Todo table.
  • results = session.exec(statement).all() executes the statement using the injected session and fetches all the resulting Todo objects.
  • The list of Todo objects is returned.

• read_todo endpoint ( @app.get("/todos/{todo_id}", response_model=TodoRead) ) :

  • This endpoint uses GET and maps to /todos/{todo_id} , where {todo_id} is a path parameter.
  • todo_id: int in the function signature tells FastAPI to expect an integer for the todo_id path parameter and parse it automatically.
  • session.get(Todo, todo_id) is a convenient SQLAlchemy method to fetch a single record by its primary key. It looks for a Todo object with the matching todo_id .
  • if todo is None: checks if a Todo item with the given ID was found. If not, it raises an HTTPException with a 404 Not Found status code and a detail message.
  • If found, the todo object is returned.

These endpoints provide the core functionality for retrieving data. You can test them by sending GET requests to /todos/ (to get all) and /todos/{id} (to get a specific one, e.g., /todos/1 ).

Implementing the Update Operation

Moving on, let’s implement the Update functionality for our FastAPI SQLModel CRUD example . This allows us to modify existing Todo items. We’ll add another endpoint to main.py .

@app.put("/todos/{todo_id}", response_model=TodoRead)
def update_todo(
    todo_id: int,
    todo_update: TodoUpdate,
    session: Session = Depends(get_session)
):
    db_todo = session.get(Todo, todo_id)
    if db_todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")

    # Get data from the update model, ignoring unset fields
    update_data = todo_update.model_dump(exclude_unset=True)

    # Update the database object with new data
    for key, value in update_data.items():
        setattr(db_todo, key, value)

    session.add(db_todo)
    session.commit()
    session.refresh(db_todo)
    return db_todo

Let’s walk through this update endpoint:

• @app.put("/todos/{todo_id}", response_model=TodoRead) :
  • This uses the PUT HTTP method, typically used for full resource replacement or updates, targeting /todos/{todo_id} .
  • response_model=TodoRead ensures the updated item is returned in the correct format.
• def update_todo(todo_id: int, todo_update: TodoUpdate, session: Session = Depends(get_session)) :
  • todo_id: int : The ID of the todo item to update.
  • todo_update: TodoUpdate : The request body, validated against our TodoUpdate model. Notice that all fields in TodoUpdate are optional, which is key for partial updates.
  • session : The database session is injected as usual.
• db_todo = session.get(Todo, todo_id) : We first fetch the existing Todo item from the database using its ID.
• if db_todo is None: : If the todo isn’t found, we return a 404 error.
• update_data = todo_update.model_dump(exclude_unset=True) : This is a powerful feature of Pydantic (which SQLModel uses). model_dump() converts the TodoUpdate Pydantic model instance into a dictionary. exclude_unset=True is crucial here; it means only fields that were actually provided in the request body will be included in this dictionary. Fields that were not sent will be excluded.
• for key, value in update_data.items(): setattr(db_todo, key, value) : We iterate through the update_data dictionary. For each key-value pair (e.g., ('completed', True) ), setattr(db_todo, key, value) dynamically sets the corresponding attribute on our db_todo object. This allows for partial updates without needing to explicitly check which fields were sent.
• session.add(db_todo) , session.commit() , session.refresh(db_todo) : These lines save the changes to the database, commit the transaction, and refresh the object to reflect the updated state.
• return db_todo : The updated Todo object is returned.

This PUT endpoint allows clients to send only the fields they want to change, making it very flexible. You can test this by sending a PUT request to /todos/{id} with a JSON body containing the fields you wish to update, e.g., {"completed": true} .

Implementing the Delete Operation

Finally, let’s complete our FastAPI SQLModel CRUD example by implementing the Delete operation. This is how we remove Todo items from our database.

Add this endpoint to main.py :

@app.delete("/todos/{todo_id}", status_code=204)
def delete_todo(todo_id: int, session: Session = Depends(get_session)):
    db_todo = session.get(Todo, todo_id)
    if db_todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")

    session.delete(db_todo)
    session.commit()
    # No refresh needed after delete, and no content to return
    return

Let’s break down the delete endpoint:

• @app.delete("/todos/{todo_id}", status_code=204) :
  • This uses the DELETE HTTP method for removing a resource, targeting /todos/{todo_id} .
  • status_code=204 : This is the standard HTTP status code for successful deletion where there is no content to return ( 204 No Content ).
• def delete_todo(todo_id: int, session: Session = Depends(get_session)) :
  • todo_id: int : The ID of the todo item to delete.
  • session : The database session is injected.
• db_todo = session.get(Todo, todo_id) : We fetch the Todo item to be deleted.
• if db_todo is None: : If the item doesn’t exist, we return a 404 error.
• session.delete(db_todo) : This marks the db_todo object for deletion within the session.
• session.commit() : This executes the deletion in the database.
• return : Since we specified status_code=204 , we don’t return any body content. An empty return signifies success.

This completes our basic CRUD operations. You can test this by sending a DELETE request to /todos/{id} .

Conclusion

And there you have it, guys! We’ve successfully built a FastAPI SQLModel CRUD example . We covered setting up our environment, defining elegant data models with SQLModel, establishing database connections, and implementing all four fundamental CRUD operations: Create, Read, Update, and Delete. The combination of FastAPI’s speed and ease of use with SQLModel’s powerful data modeling capabilities makes building database-driven APIs a breeze. You’ve seen how SQLModel simplifies things by allowing you to define models that work for both your API and your database, and how FastAPI’s dependency injection makes managing database sessions clean and efficient. This example is a solid foundation, and you can build upon it to create more complex applications. Keep experimenting, keep coding, and happy building!