Connecting FastAPI to a Database with SQLModel

8 min read

By David Muraya • August 26, 2025

When building an API with FastAPI, you'll often need to connect to a SQL database. While you can use any database library, SQLModel was created by the same author as FastAPI to be a perfect match. It combines SQLAlchemy and Pydantic into a single, easy-to-use library.

Because SQLModel is built on SQLAlchemy, it supports any database that SQLAlchemy does, including PostgreSQL, MySQL, SQLite, Oracle, and Microsoft SQL Server.

This guide will walk you through setting up a database connection, creating models, and building a complete CRUD (Create, Read, Update, Delete) application with FastAPI and SQLModel.

1. Creating Your Database Model

A model is a Python class that defines the structure of a database table. With SQLModel, you create a class that inherits from SQLModel.

Here’s an example of a Product model, which you can place in a file like app/services/database/models.py.

# In app/services/database/models.py
from typing import Optional

from sqlmodel import Field, SQLModel

class Product(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    price: float
    description: Optional[str] = None

This looks a lot like a Pydantic model, but with a few key differences:

• table=True tells SQLModel that this class represents a database table, not just a data validation model.
• Field(primary_key=True) marks the id field as the table's primary key. We define it as Optional[int] because the database will generate the ID for us when we create a new product.
• Field(index=True) tells SQLModel to create a database index on the name column, which makes searching by name faster.
• A standard str type is mapped to a TEXT or VARCHAR column in the database.

2. Setting Up the Database Connection

To communicate with the database, you need an "engine." The engine is the central point of contact. You create it once for your entire application.

Let's set this up in a session.py file. This example uses SQLite, which is great for local development because it uses a single file for the database.

# In a file like app/services/database/session.py
from pathlib import Path

from sqlmodel import Session, create_engine

from app.config.main import get_settings

settings = get_settings()

DATABASE_PATH = Path(settings.DB)
DATABASE_URL = f"sqlite:///{DATABASE_PATH}"
CONNECT_ARGS = {"check_same_thread": False}

engine = create_engine(DATABASE_URL, connect_args=CONNECT_ARGS)

def get_session():
    """Yields a SQLModel Session instance."""
    with Session(engine) as session:
        yield session

Here, we create an engine using our database URL. The get_session function is a dependency that we'll use in our API routes to get a database session. A session is a short-lived workspace for all your database operations. The yield statement provides the session to the route and ensures it's properly closed afterward, even if an error occurs.

For more on managing settings with a .env file, check out my guide on Centralizing Your FastAPI Configuration Settings.

3. Creating Tables on App Startup

FastAPI allows you to run code when the application starts up. We can use this on_startup event to create our database tables based on our SQLModel models.

First, create a function to handle table creation. This can live with your models.

# In a file like app/services/database/models.py
from .session import engine

def configure():
    SQLModel.metadata.create_all(bind=engine)

Then, call this function from an on_startup event handler in your main.py:

# In app/config/events.py
from app.services.database.models import configure

async def on_start_up():
    configure()

app = FastAPI(on_startup=[on_start_up])

Now, when your FastAPI application starts, it will connect to the database and create the product table if it doesn't already exist.

4. Creating the Product Schemas

Before writing the CRUD functions, define the schemas that will be used for input validation and API responses. These schemas help separate what data is required for creating, reading, updating, or deleting a product.

# In a file like app/schemas/product_models.py
from typing import Optional

from sqlmodel import SQLModel

class ProductBase(SQLModel):
    name: str
    price: float
    description: Optional[str] = None

class ProductCreate(ProductBase):
    pass

class ProductRead(ProductBase):
    id: int

class ProductUpdate(SQLModel):
    name: Optional[str] = None
    price: Optional[float] = None
    description: Optional[str] = None

class ProductDelete(SQLModel):
    id: int

• ProductBase defines the common fields for a product.
• ProductCreate is used when creating a new product.
• ProductRead is used for responses and includes the product ID.
• ProductUpdate allows partial updates, so all fields are optional.
• ProductDelete can be used for delete operations if needed.

5. Creating the CRUD Functions

To keep our code organized, we'll separate the database logic from the API endpoint logic. We'll create a set of functions to handle the actual database operations for our Product model.

# In a file like app/services/database/crud_product.py
from typing import List, Optional

from sqlmodel import Session, select

from app.schemas.product_models import ProductCreate, ProductUpdate
from app.services.database.models import Product


async def create_product(session: Session, product_create: ProductCreate) -> Product:
    product = Product.model_validate(product_create)
    session.add(product)
    return product


async def get_product(session: Session, product_id: int) -> Optional[Product]:
    return session.get(Product, product_id)


async def get_products(
    session: Session, skip: int = 0, limit: int = 100
) -> List[Product]:
    statement = select(Product).offset(skip).limit(limit)
    result = session.exec(statement)
    return result.all()


async def update_product(
    session: Session, product_id: int, product_update: ProductUpdate
) -> Optional[Product]:
    product = await get_product(session, product_id)
    if not product:
        return None
    update_data = product_update.model_dump(exclude_unset=True)
    for key, value in update_data.items():
        setattr(product, key, value)
    session.add(product)
    return product


async def delete_product(session: Session, product_id: int) -> bool:
    product = await get_product(session, product_id)
    if not product:
        return False
    session.delete(product)
    return True

These functions handle the direct interaction with the database session:

• create_product: Creates a Product instance from the input data and adds it to the session.
• get_product: Retrieves a single product by its ID using session.get().
• get_products: Fetches a list of products with pagination.
• update_product: Finds a product, updates its fields, and adds the updated object to the session.
• delete_product: Finds a product and marks it for deletion in the session.

6. Building the CRUD Endpoints

Now we can build the API endpoints that call our new CRUD functions. We'll use the get_session dependency to get a database session in each route.

Here is a complete main.py with all the CRUD operations:

# In main.py
from typing import List

from fastapi import Depends, FastAPI, HTTPException, Query, Response, status
from sqlmodel import Session

from app.config.events import on_shutdown, on_start_up
from app.services.database.crud_product import (
    create_product,
    delete_product,
    get_product,
    get_products,
    update_product,
)
from app.services.database.session import get_session
from app.schemas.product_models import ProductCreate, ProductRead, ProductUpdate

app = FastAPI(on_shutdown=[on_shutdown], on_startup=[on_start_up])

@app.post("/products/", response_model=ProductRead)
async def create_product_endpoint(product: ProductCreate, db: Session = Depends(get_session)):
    db_product = await create_product(db, product)
    db.commit()
    db.refresh(db_product)
    return db_product

@app.get("/products/{product_id}", response_model=ProductRead)
async def read_product_endpoint(product_id: int, db: Session = Depends(get_session)):
    db_product = await get_product(db, product_id)
    if not db_product:
        raise HTTPException(status_code=404, detail="Product not found")
    return db_product

@app.get("/products/", response_model=List[ProductRead])
async def read_products_endpoint(
    skip: int = 0, limit: int = 100, db: Session = Depends(get_session)
):
    return await get_products(db, skip=skip, limit=limit)

@app.patch("/products/{product_id}", response_model=ProductRead)
async def update_product_endpoint(product_id: int, product: ProductUpdate, db: Session = Depends(get_session)):
    db_product = await update_product(db, product_id, product)
    if not db_product:
        raise HTTPException(status_code=404, detail="Product not found")
    db.commit()
    db.refresh(db_product)
    return db_product

@app.delete("/products/{product_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_product_endpoint(product_id: int, db: Session = Depends(get_session)):
    deleted = await delete_product(db, product_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="Product not found")
    db.commit()
    return Response(status_code=status.HTTP_204_NO_CONTENT)

In the create_product_endpoint, for example:

1. It receives product data that conforms to the ProductCreate model.
2. It uses Depends(get_session) to get a database session.
3. It calls the create_product function.
4. db.commit() saves the changes to the database.
5. db.refresh(db_product) updates the db_product object with the new ID generated by the database.

You can implement middleware and logging on your fastapi application in this main.py. Learn more in my articles on Essential FastAPI Middlewares and How to Set Up Logging in FastAPI.

7. Handling Serverless Databases

If you're using a serverless database like Neon that can scale to zero, connections can go stale. To prevent errors, you should use pool_pre_ping=True when creating your engine. This tells SQLAlchemy to check the connection before using it.

Read more in my guide on Connecting FastAPI to a Serverless Database with pool_pre_ping.

8. Conclusion

SQLModel provides a clean and efficient way to work with SQL databases in FastAPI. By defining your models once and using FastAPI's dependency injection system for sessions, you can build robust and maintainable CRUD APIs with minimal boilerplate code.

---

FAQ

Q: What's the difference between SQLModel and SQLAlchemy? A: SQLModel is a library that sits on top of SQLAlchemy and Pydantic. It combines the features of both, giving you a single class for your database model, data validation, and API schema, which simplifies your code.

Q: Do I have to use SQLModel with FastAPI? A: No. You can use SQLAlchemy directly, or other ORMs like Tortoise ORM or Pony ORM. However, SQLModel is designed to integrate seamlessly with FastAPI's patterns.

Q: Why use yield in get_session? A: Using yield creates a generator. In FastAPI's dependency system, this allows code to run before the response is sent (providing the session) and after the response is sent (closing the session). This ensures database connections are always cleaned up properly.

Q: What do db.commit() and db.refresh() do? A: db.commit() saves all the changes you've made in the current session (like adding, updating, or deleting objects) to the database. db.refresh(obj) updates a Python object with the latest data from the database, which is useful for getting auto-generated values like an id.

Q: Why don't we commit the transaction inside the CRUD function? A: Committing the transaction in the API route, rather than inside the CRUD function, gives you more control. This way, you can group multiple operations into a single transaction if needed, handle errors or rollbacks at the API level, and keep your CRUD functions focused on database logic only. It also makes your code easier to test and maintain, since the side effect of committing is explicit and not hidden inside a helper function.

Q: How do I handle database migrations? A: For managing changes to your database schema over time, you should use a migration tool like Alembic. You can read my guide on Running Database Migrations with Alembic in Google Cloud Build to get started.