Essential FastAPI Guide: Creating a CRUD API with SQLModel and Database Integration

In this tutorial, we will be building a CRUD API using FastAPI with a database and SQLModel. FastAPI is a modern web framework for building APIs with Python, while SQLModel is a Python library that helps to define and interact with SQL databases using standard Python data structures.

For this tutorial, we will be using SQLite as our database, but you can easily switch to another database by changing the database URL in the connection string.

Let’s get started by setting up our environment and installing the necessary libraries.

1. Create a new directory for our project and navigate to it in the terminal.

   mkdir fastapi_crud
   cd fastapi_crud

2. Create a virtual environment and activate it.

   python3 -m venv venv
   source venv/bin/activate

3. Install FastAPI and other required libraries.

   pip install fastapi uvicorn sqlalchemy databases[sqlite] sqlmodel

4. Create a new file named main.py in your project directory and import the necessary libraries.

   from fastapi import FastAPI, HTTPException
   from fastapi.responses import JSONResponse
   from sqlalchemy import create_engine
   from databases import Database
   from sqlmodel import SQLModel, Field, Session, create_engine as db_create_engine

5. Next, let’s define our database models using SQLModel. We will create a simple User model with id, name, and email fields.

   class User(SQLModel, table=True):
   id: int = Field(default=None, primary_key=True)
   name: str
   email: str

6. Now, let’s set up our database connection using SQLAlchemy and databases.

   
   DATABASE_URL = "sqlite:///./test.db"

database = Database(DATABASE_URL)
engine = create_engine(DATABASE_URL)
session = Session(engine)

7. Let's create our FastAPI application and define our CRUD endpoints for the `User` model.
```python
app = FastAPI()

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

@app.post("/users/", response_model=User)
async def create_user(user: User):
    query = User.insert().values(**user.dict())
    user.id = await database.execute(query)
    return user

@app.get("/users/{user_id}", response_model=User)
async def get_user(user_id: int):
    query = User.select().where(User.id == user_id)
    user = await database.fetch_one(query)
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.put("/users/{user_id}", response_model=User)
async def update_user(user_id: int, user: User):
    query = User.update().where(User.id == user_id).values(**user.dict())
    await database.execute(query)
    updated_user = await database.fetch_one(User.select().where(User.id == user_id))
    if updated_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return updated_user

@app.delete("/users/{user_id}", response_model=User)
async def delete_user(user_id: int):
    query = User.delete().where(User.id == user_id)
    deleted_user = await database.fetch_one(User.select().where(User.id == user_id))
    if deleted_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    await database.execute(query)
    return deleted_user

@app.get("/users/", response_model=list[User])
async def list_users():
    query = User.select()
    users = await database.fetch_all(query)
    return users

1. Finally, let’s start our FastAPI application using Uvicorn.

   uvicorn main:app --reload

You can now test your CRUD API by navigating to http://localhost:8000/docs in your browser. You should see the Swagger UI where you can interact with your API by creating, reading, updating, and deleting users.

That’s it! You have successfully built a CRUD API using FastAPI with a database and SQLModel. You can extend this example by adding more models, endpoints, and custom business logic to suit your specific requirements.