Building A Simple CRUD Application With FastAPI

2021-10-05 5065 words 24 minutes

See the code for this project on GitHub.

Intro

In this tutorial we’ll build a very simple “To Do” list application with FastAPI. Users will be able to

Create To Do list items
Read To Do list items
Update To Do list items
Delete To Do list items

Create. Read. Update. Delete. “CRUD”.

We’re going to build a backend application. So, a REST API with a database only. (In theory, you could build a front-end on top of it with something like React or Angular.)

Let’s begin.

Setup

If you’re brand new to FastAPI, check out my tutorial on implementing Rock Paper Scissors.

We’ll call our application todooo - all lowercase to keep it hip. I’ll start this project the same way I do all fastapi projects, by

Creating a new python virtual environment (for me, conda create -n py39 python=3.9)
Installing fastapi and uvicorn (for me, pip install 'fastapi[all]')
Setting up a root project directory called todooo/
Setting up a new PyCharm project
Initializing a git repo and a .gitignore file

At this point my project file structure looks like this

todoo/
├── .git/
├── .gitignore
├── .idea/
└── README.md

and my .gitignore file looks like this

.gitignore

.DS_Store
.idea/
__pycache__/

The next thing I’ll do is create a main.py file inside todooo/ with some boilerplate code just to get my app up and running.

main.py

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
def root():
    return {"message": "Hello World"}

I can run this app from Terminal with uvicorn main:app --reload and view it locally at http://127.0.0.1:8000/. At this point, the root page of my app simply returns {"message":"Hello World"}

And of course I can browse the auto-generated Swagger docs for my app at http://127.0.0.1:8000/docs.

Designing The App

Now let’s put some thought into our app. What do we want it to do (heh)? As a user I want to be able to create, read, update, and delete todo list items. So, we should probably have five endpoints like

functionality
create a todo item
read a todo list item
update a todo item
delete a todo item
read all todo items

The type of operation we’re performing dictates which HTTP method or “verb” we should use. For example, our endpoint for reading a todo list item should use an HTTP GET method. Our endpoint for deleting a todo list item should use an HTTP DELETE method.

functionality	method
create a todo item	POST
read a todo list item	GET
update a todo item	PUT
delete a todo item	DELETE
read all todo items	GET

A quick note about POST vs PUT… Generally speaking, POST is used to a create a new database record and PUT is used to update an existing database record. However, if you want to create a new database record and you know the unique identifier ahead of time, you should probably use PUT. In our case, when a user creates a new todo item, the unique identifier for that item should be created automatically by the database, not provided by the user. Thus, POST is the appropriate method for us to create a new todo item. (See here for a lengthier discussion on the topic.)

What should the URL paths for these operations be? How about the following

functionality	method	path
create a todo item	POST	/todo
read a todo list item	GET	/todo/{id}
update a todo item	PUT	/todo/{id}
delete a todo item	DELETE	/todo/{id}
read all todo items	GET	/todo

Yes - we can actually implement different logic for the same URL path, depending on the HTTP method used. A PUT call to todooo.com/todo/123 will implement different logic than a DELETE call to todooo.com/todo/123. By the way, the combination of URL path and HTTP method is called a “path operation” in FastAPI.

Dummy App

Cool. Now lets implement some path operation functions for our desired endpoints. We’ll fill them with dummy, starter code for now.

main.py

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def root():
    return "todooo"

@app.post("/todo")
def create_todo():
    return "create todo item"

@app.get("/todo/{id}")
def read_todo(id: int):
    return "read todo item with id {id}"

@app.put("/todo/{id}")
def update_todo(id: int):
    return "update todo item with id {id}"

@app.delete("/todo/{id}")
def delete_todo(id: int):
    return "delete todo item with id {id}"

@app.get("/todo")
def read_todo_list():
    return "read todo list"

Visiting our auto-generated Swagger docs, we can see and play with our API.

There’s something bad we should improve. Did you catch it? If we expand the POST /todo endpoint we can see it generates a 200 SUCCESSFUL RESPONSE status code.

A 200 response is “okay” here in the same way, when I ask my wife to rate the steaks I grilled and she says “they’re okay..”. A better positive response here would be to issue a 201 CREATED response (this is of course assuming the todo item was successfully created). We can do this by

importing the status module from fastapi: from fastapi import FastAPI, status
adding the status_code=status.HTTP_201_CREATED to the path operation decorator: @app.post("/todo", status_code=status.HTTP_201_CREATED)

So our updated main.py file looks like this

from fastapi import FastAPI, status

app = FastAPI()

@app.get("/")
def root():
    return "todooo"

@app.post("/todo", status_code=status.HTTP_201_CREATED)
def create_todo():
    return "create todo item"

@app.get("/todo/{id}")
def read_todo(id: int):
    return "read todo item with id {id}"

@app.put("/todo/{id}")
def update_todo(id: int):
    return "update todo item with id {id}"

@app.delete("/todo/{id}")
def delete_todo(id: int):
    return "delete todo item with id {id}"

@app.get("/todo")
def read_todo_list():
    return "read todo list"

The Database

Next let’s incorporate a sqlite database to store our todo list items. The workhorse package we’ll use for database operations is sqlalchemy. sqlalchemy is powerful but complex. I’ll probably write an intro post about it soon..

First, let’s install sqlalchemy with pip install sqlalchemy.

SqlAlchemy Engine

Then we’ll need to create an Engine instance. An Engine tells sqlalchemy

what type of database you’re connecting to
where that database is located
what your credentials are for connecting to the database
lots of other stuff

To create an engine we use the create_engine() function like this.

from sqlalchemy import create_engine

engine = create_engine("sqlite:///todooo.db")

(This’ll go in main.py as shown below.)

The first (and only required) parameter inside create_engine() is the database url. Typically it takes the form dialect+driver://username:password@host:port/database. In our case, dialect is sqlite hence the sqlite:// bit. The additional /todooo.db bit specifies the location of the database. Of course, our sqlite database hasn’t been created yet, but this is where it will be. Note that this path is relative to the working directory.

In a production setting, the call to create_engine() might look more like engine = create_engine('postgresql+psycopg2://scott:tiger@localhost/mydatabase') or engine = create_engine('mysql://scott:tiger@localhost/foo'). See here for more details.

Every Engine has a Dialect. The Dialect refers to the flavor of SQL in the database (e.g. sqlite, postgres, mysql, etc.). If you inspect engine.dialect you’ll see something like <sqlalchemy.dialects.sqlite.pysqlite.SQLiteDialect_pysqlite object at 0x7ff6180eddc0>. Part of the engine’s job is to transform python objects and sqlalchemy functions into SQL code that can be interpreted by the database. Keep in mind that different flavors of SQL use different syntax. In mysql, foo || bar translates to “foo or bar” but in postgres it translates to the concatenation “foobar”. So, it’s not enough to simply “translate python code into SQL code”. The engine has to “translate python code in sqlite code” or “python code into mysql code” or “python code into postgresql code”.

SqlAlchemy Declarative

The next thing we need to do is create a Declarative. This is where things start to get hella confusing.. Let me start by showing you the code

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine("sqlite:///todooo.db")
Base = declarative_base()

declarative_base() here is a factory function. What it spits out is a metaclass - basically just a nother Python class we can use. It’s purpose will become more clear later.

Database Initialization

The next thing we need to do is initialize the database, which we’ll do with Base.metadata.create_all(engine). We’ll place all this database code in our main.py file before we initialize our app with app = FastAPI(). (After all, we only want to intialize the database once - not every time someone interacts with our application.)

main.py

from fastapi import FastAPI, status
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base

# Create a sqlite engine instance
engine = create_engine("sqlite:///todooo.db")

# Create a DeclarativeMeta instance
Base = declarative_base()

# Create the database
Base.metadata.create_all(engine)

# Initialize app
app = FastAPI()

@app.get("/")
def root():
    return "todooo"

@app.post("/todo", status_code=status.HTTP_201_CREATED)
def create_todo():
    return "create todo item"

@app.get("/todo/{id}")
def read_todo(id: int):
    return "read todo item with id {id}"

@app.put("/todo/{id}")
def update_todo(id: int):
    return "update todo item with id {id}"

@app.delete("/todo/{id}")
def delete_todo(id: int):
    return "delete todo item with id {id}"

@app.get("/todo")
def read_todo_list():
    return "read todo list"

If we run our app with uvicorn main:app --reload, you’ll notice a new todooo.db file gets generated inside our project. This is our sqlite database.

todooo/
├── .git/
├── .gitignore
├── .idea/
├── README.md
├── __pycache__/
├── main.py
└── todooo.db

Table Creation

First off, what database tables do we need for our app? Well, in this case one table will suffice; a todos table like

id	task
1	cut the grass
2	buy anniversary gift
3	clean the fridge
4	live
5	love
6	laugh

So, two columns - id and task where id is the primary key (your typical auto incrementing integer field) and task is a string field. How do we make such a table in sqlalchemy? The answer is to build a Python class like this

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

# Create a sqlite engine instance
engine = create_engine("sqlite:///todooo.db")

# Create a DeclarativeMeta instance
Base = declarative_base()

# Define To Do class inheriting from Base
class ToDo(Base):
    __tablename__ = 'todos'
    id = Column(Integer, primary_key=True)
    task =  Column(String(50))

# Create the database
Base.metadata.create_all(engine)

Here we

Declare a ToDo class that inherits from Base. Our Todo class will correspond to a todos table in the database, where each instance of our Todo class will correspond to a row in the todos table.
Give ToDo three attributes:
1. __tablename__ = 'todos' sets the name of the table in the database as “todos”
2. id = Column(Integer, primary_key=True) sets the primary key of the table as an integer column named “id” (which by default is auto incrementing)
3. task = Column(String(256)) creates a string field (up to 256 characters) in the todos table named “task”
Initialize the tables with Base.metadata.create_all(engine)

Why inherit from Base?
Base is like a ledger. By inheriting from it, it records that we want a todo table, and every time we create a new todo instance (shown later) Base records that we want a new todo record in the todo table.

Again, we’ll just embed the above code in our main.py file.

main.py

from fastapi import FastAPI, status
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

# Create a sqlite engine instance
engine = create_engine("sqlite:///todooo.db")

# Create a DeclarativeMeta instance
Base = declarative_base()

# Define To Do class inheriting from Base
class ToDo(Base):
    __tablename__ = 'todos'
    id = Column(Integer, primary_key=True)
    task = Column(String(256))

# Create the database
Base.metadata.create_all(engine)

# Initialize app
app = FastAPI()

@app.get("/")
def root():
    return "todooo"

@app.post("/todo", status_code=status.HTTP_201_CREATED)
def create_todo():
    return "create todo item"

@app.get("/todo/{id}")
def read_todo(id: int):
    return "read todo item with id {id}"

@app.put("/todo/{id}")
def update_todo(id: int):
    return "update todo item with id {id}"

@app.delete("/todo/{id}")
def delete_todo(id: int):
    return "delete todo item with id {id}"

@app.get("/todo")
def read_todo_list():
    return "read todo list"

After we run our app with uvicorn main:app --reload, a todos table gets created inside the todooo.db file.

Database Code Refactor

Our main.py file is getting messy. Let’s put the database logic into a database.py file within the project root directory, todooo/ and then import it from main.py. So, our project structure now looks like this

todooo/
├── .git/
├── .gitignore
├── .idea/
├── README.md
├── __pycache__/
├── database.py
└── main.py

database.py

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

# Create a sqlite engine instance
engine = create_engine("sqlite:///todooo.db")

# Create a DeclarativeMeta instance
Base = declarative_base()

# Define To Do class inheriting from Base
class ToDo(Base):
    __tablename__ = 'todos'
    id = Column(Integer, primary_key=True)
    task = Column(String(256))

main.py

from fastapi import FastAPI, status
from database import Base, engine

# Create the database
Base.metadata.create_all(engine)

# Initialize app
app = FastAPI()

@app.get("/")
def root():
    return "todooo"

@app.post("/todo", status_code=status.HTTP_201_CREATED)
def create_todo():
    return "create todo item"

@app.get("/todo/{id}")
def read_todo(id: int):
    return "read todo item with id {id}"

@app.put("/todo/{id}")
def update_todo(id: int):
    return "update todo item with id {id}"

@app.delete("/todo/{id}")
def delete_todo(id: int):
    return "delete todo item with id {id}"

@app.get("/todo")
def read_todo_list():
    return "read todo list"

CRUD Operations

Now let’s implement some crap! I mean, CRUD.

Create

First we need to implement the logic for creating a new todo item. We expect the user to input one thing - the task string. Since object creations happen via POST, data coming from the user should be sent in the body of the HTTP request.

To accomplish this, we need to declare a ToDo pydantic model which looks like this

from pydantic import BaseModel

# Create ToDoRequest Base Model
class ToDoRequest(BaseModel):
    task: str

This is not to be confused with our ToDo database model (in database.py) we created earlier. The ToDo database model tells the database how to set up the todos table. The ToDo pydantic model defines the acceptable input data format for creating a new todo item. There’s a difference - specifically, when a user creates a new todo item, they should not specify the id since it’s automatically created by the database.

We’ll place this code in main.py for now.

from fastapi import FastAPI, status
from database import Base, engine
from pydantic import BaseModel

# Create ToDoRequest Base Model
class ToDoRequest(BaseModel):
    task: str

# Create the database
Base.metadata.create_all(engine)

# Initialize app
app = FastAPI()

@app.get("/")
def root():
    return "todooo"

@app.post("/todo", status_code=status.HTTP_201_CREATED)
def create_todo():
    return "create todo item"

@app.get("/todo/{id}")
def read_todo(id: int):
    return "read todo item with id {id}"

@app.put("/todo/{id}")
def update_todo(id: int):
    return "update todo item with id {id}"

@app.delete("/todo/{id}")
def delete_todo(id: int):
    return "delete todo item with id {id}"

@app.get("/todo")
def read_todo_list():
    return "read todo list"

Then we’ll change

@app.post("/todo", status_code=status.HTTP_201_CREATED)
def create_todo():
    return "create todo item"

@app.post("/todo", status_code=status.HTTP_201_CREATED)
def create_todo(todo: ToDoRequest):
    # insert todo as a new record in the database
    return "create todo item"

Now our API should be “aware” that something resembling a ToDoRequest shoud be passed into the body of the HTTP request for the POST /todo endpoint. Let’s see how this looks in the swagger docs for our app.

Notice our POST /todo endoint now has a prefilled JSON string with one field: task. (I changed the default value to “take out the trash”). Now I’ll hit execute.

We received a 201 response with a response body “create todo item”. Looks like it’s working properly.. Now we need to actually insert the data into the database. Here’s some starter code to make that work.

from fastapi import FastAPI, status
from database import Base, engine, ToDo
from pydantic import BaseModel
from sqlalchemy.orm import Session

# Create ToDoRequest Base Model
class ToDoRequest(BaseModel):
    task: str

# Create the database
Base.metadata.create_all(engine)

# Initialize app
app = FastAPI()

# <irrelevant code here..>

@app.post("/todo", status_code=status.HTTP_201_CREATED)
def create_todo(todo: ToDoRequest):

    # create a new database session
    session = Session(bind=engine, expire_on_commit=False)

    # create an instance of the ToDo database model
    tododb = ToDo(task = todo.task)

    # add it to the session and commit it
    session.add(tododb)
    session.commit()

    # grab the id given to the object from the database
    id = tododb.id

    # close the session
    session.close()

    # return the id
    return f"created todo item with id {id}"

# <irrelevant code here..>

Notes

We need to import ToDo from database and Session from sqlalchemy.orm
Inside create_todo(), we start by initializing a new database session. A session facilitates the exchange of data between the database and our API. Don’t worry about the details for now. (That’s a topic for another day.)
We use the data from the request (i.e. from the ToDoRequest object, todo) to create a ToDo database model instance which we’ll call tododb.
We add tododb to the session and commit it.
The session automatically updates our tododb object with the database-generated id.
We return the id in the response string.

There are a lot of ways we can improve this code, but we’ll refactor it later on. Also, I recently discovered the TablePlus app which seems like a nice way to view and interact with a database. After creating some todo items and connecting the TablePlus app to my todooo.db file, I can view them like so

Now let’s move onto the RUD operations.

Read

We need to implement two different types of reads:

read_todo(id: int) for reading a single, specific todo item and
read_todo_list(): for reading all todo items

We start with read_todo(id: int).

@app.get("/todo/{id}")
def read_todo(id: int):

    # create a new database session
    session = Session(bind=engine, expire_on_commit=False)

    # get the todo item with the given id
    todo = session.query(ToDo).get(id)

    # close the session
    session.close()

    return f"todo item with id: {todo.id} and task: {todo.task}"

If we try it out, it works :)

..unless we try to read a todo item with a non existent id :(

Obviously we need to better handle the case where the user requests a todo item id that doesn’t exist. Handling this case is pretty simple thanks to FastAPI’s HTTPException module. It looks like this

from fastapi import HTTPException
# <other imports not shown here>

@app.get("/todo/{id}")
def read_todo(id: int):

    # create a new database session
    session = Session(bind=engine, expire_on_commit=False)

    # get the todo item with the given id
    todo = session.query(ToDo).get(id)

    # close the session
    session.close()

    return f"todo item with id: {todo.id} and task: {todo.task}"

Now when we request item with id 999, we get a 404 error with useful feedback.

One last thing I want to clean up before moving on.. When the use requests a todo item that exists, we’re returning a string like ‘todo item with id: 1 and task: “buy food”’. A better response would be to return the todo object serialized as JSON, like this

{
  "task": "buy food",
  "id": 1
}

To make that happen, we can simply return the todo object we fetched from the database. In other words, we can change return f"todo item with id: {todo.id} and task: {todo.task}" to return todo. (See here for details on customizing the response.)

At this point, our main.py file looks like this.

from fastapi import FastAPI, status, HTTPException
from database import Base, engine, ToDo
from pydantic import BaseModel
from sqlalchemy.orm import Session

# Create ToDoRequest Base Model
class ToDoRequest(BaseModel):
    task: str

# Create the database
Base.metadata.create_all(engine)

# Initialize app
app = FastAPI()

@app.get("/")
def root():
    return "todooo"

@app.post("/todo", status_code=status.HTTP_201_CREATED)
def create_todo(todo: ToDoRequest):

    # create a new database session
    session = Session(bind=engine, expire_on_commit=False)

    # create an instance of the ToDo database model
    tododb = ToDo(task = todo.task)

    # add it to the session and commit it
    session.add(tododb)
    session.commit()

    # grab the id given to the object from the database
    id = tododb.id

    # close the session
    session.close()

    # return the id
    return f"created todo item with id {id}"

@app.get("/todo/{id}")
def read_todo(id: int):

    # create a new database session
    session = Session(bind=engine, expire_on_commit=False)

    # get the todo item with the given id
    todo = session.query(ToDo).get(id)

    # close the session
    session.close()

    # check if todo item with given id exists. If not, raise exception and return 404 not found response
    if not todo:
        raise HTTPException(status_code=404, detail=f"todo item with id {id} not found")

    return todo

@app.put("/todo/{id}")
def update_todo(id: int):
    return "update todo item with id {id}"

@app.delete("/todo/{id}")
def delete_todo(id: int):
    return "delete todo item with id {id}"

@app.get("/todo")
def read_todo_list():
    return "read todo list"

Now let’s see how to fetch all todo items. Before we do, what should the response even look like? The answer is a nested collection of todo items. So, just as one todo item is serialized like

{
  "task": "buy food",
  "id": 1
}

three todo items should get serialized like

[
  {
    "task": "buy food",
    "id": 1
  },
  {
    "task": "buy groceries",
    "id": 2
  },
  {
    "task": "walk the dog",
    "id": 3
  }
]

Fortunately, FastAPI handles the serialization for us. All we need to do is fetch the data. So, a simple implementation of this goes as follows.

@app.get("/todo")
def read_todo_list():
    # create a new database session
    session = Session(bind=engine, expire_on_commit=False)

    # get all todo items
    todo_list = session.query(ToDo).all()

    # close the session
    session.close()

    return todo_list

There’s lots of room for improvement here, but let’s carry on.

Update

Now let’s write the logic to update a todo item.

@app.put("/todo/{id}")
def update_todo(id: int, task: str):

    # create a new database session
    session = Session(bind=engine, expire_on_commit=False)

    # get the todo item with the given id
    todo = session.query(ToDo).get(id)

    # update todo item with the given task (if an item with the given id was found)
    if todo:
        todo.task = task
        session.commit()

    # close the session
    session.close()

    # check if todo item with given id exists. If not, raise exception and return 404 not found response
    if not todo:
        raise HTTPException(status_code=404, detail=f"todo item with id {id} not found")

    return todo

Notes

I added a query parameter task: str. So, users are expected to make requests like todooo.com/todo/3?task=workout
I use if todo: to check if the requested todo item even exists. If it does, I update its task attribute and update the database with session.commit()

Delete

Lastly, we need the ability to delete a todo item. This is pretty straight-forward given the stuff we’ve already implemented.

@app.delete("/todo/{id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_todo(id: int):

    # create a new database session
    session = Session(bind=engine, expire_on_commit=False)

    # get the todo item with the given id
    todo = session.query(ToDo).get(id)

    # if todo item with given id exists, delete it from the database. Otherwise raise 404 error
    if todo:
        session.delete(todo)
        session.commit()
        session.close()
    else:
        raise HTTPException(status_code=404, detail=f"todo item with id {id} not found")

    return None

Notes

I couldn’t decide what to return upon successful deletion of a todo item, so I decided not to return anything (i.e. return None). I adjusted my status_code accordingly to status_code=status.HTTP_204_NO_CONTENT.
To actually delete the requested todo item, we simply fetch it from the database and then call session.delete(todo) followed by session.commit().

Refactoring

Let’s fix our code..

Database Models vs Pydantic Models

Right now, our code doesn’t make a clear distinction between our datase ToDo model and our Pydantic ToDo model. FastAPI suggests calling Pydantic models schemas to help make the distinction clear. Accordingly, let’s put all our database models into a models.py file and all our Pydantic models into a schemas.py file. In doing this, we’ll also need to update database.py and main.py.

models.py

from sqlalchemy import Column, Integer, String
from database import Base

# Define To Do class inheriting from Base
class ToDo(Base):
    __tablename__ = 'todos'
    id = Column(Integer, primary_key=True)
    task = Column(String(256))

schemas.py

from pydantic import BaseModel

# Create ToDo Schema (Pydantic Model)
class ToDo(BaseModel):
    task: str

Note: Here we’ve changed the class name from ToDoRequest to just ToDo.

database.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base

# Create a sqlite engine instance
engine = create_engine("sqlite:///todooo.db")

# Create a DeclarativeMeta instance
Base = declarative_base()

main.py

from fastapi import FastAPI, status, HTTPException
from database import Base, engine
from sqlalchemy.orm import Session
import models
import schemas

# Create the database
Base.metadata.create_all(engine)

# Initialize app
app = FastAPI()

@app.get("/")
def root():
    return "todooo"

@app.post("/todo", status_code=status.HTTP_201_CREATED)
def create_todo(todo: schemas.ToDo):

    # create a new database session
    session = Session(bind=engine, expire_on_commit=False)

    # create an instance of the ToDo database model
    tododb = models.ToDo(task = todo.task)

    # add it to the session and commit it
    session.add(tododb)
    session.commit()

    # grab the id given to the object from the database
    id = tododb.id

    # close the session
    session.close()

    # return the id
    return f"created todo item with id {id}"

@app.get("/todo/{id}")
def read_todo(id: int):

    # create a new database session
    session = Session(bind=engine, expire_on_commit=False)

    # get the todo item with the given id
    todo = session.query(models.ToDo).get(id)

    # close the session
    session.close()

    # check if todo item with given id exists. If not, raise exception and return 404 not found response
    if not todo:
        raise HTTPException(status_code=404, detail=f"todo item with id {id} not found")

    return todo

@app.put("/todo/{id}")
def update_todo(id: int, task: str):

    # create a new database session
    session = Session(bind=engine, expire_on_commit=False)

    # get the todo item with the given id
    todo = session.query(models.ToDo).get(id)

    # update todo item with the given task (if an item with the given id was found)
    if todo:
        todo.task = task
        session.commit()

    # close the session
    session.close()

    # check if todo item with given id exists. If not, raise exception and return 404 not found response
    if not todo:
        raise HTTPException(status_code=404, detail=f"todo item with id {id} not found")

    return todo

@app.delete("/todo/{id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_todo(id: int):

    # create a new database session
    session = Session(bind=engine, expire_on_commit=False)

    # get the todo item with the given id
    todo = session.query(models.ToDo).get(id)

    # if todo item with given id exists, delete it from the database. Otherwise raise 404 error
    if todo:
        session.delete(todo)
        session.commit()
        session.close()
    else:
        raise HTTPException(status_code=404, detail=f"todo item with id {id} not found")

    return None

@app.get("/todo")
def read_todo_list():
    # create a new database session
    session = Session(bind=engine, expire_on_commit=False)

    # get all todo items
    todo_list = session.query(models.ToDo).all()

    # close the session
    session.close()

    return todo_list

Notice in main.py, since we import models and import schemas we reference every ToDo class like models.ToDo or schemas.ToDo. Now it’s clear in the code where we’re using a database model and where we’re using a pydantic model (i.e. schema).

Declaring Response Models

Check out the docs for our GET todo/{id} endpoint.

It indicates that a successful 200 response will return a string. But that’s not true.. It’ll actually return a JSON representation of the fetched todo item. The reason the docs are wrong here is because we haven’t specified a response_model in the @app.get() path operation decorator. Let’s do that by simply adding the parameter response_model=schemas.ToDo.

@app.get("/todo/{id}", response_model=schemas.ToDo)
def read_todo(id: int):

    # create a new database session
    session = Session(bind=engine, expire_on_commit=False)

    # get the todo item with the given id
    todo = session.query(models.ToDo).get(id)

    # close the session
    session.close()

    # check if todo item with given id exists. If not, raise exception and return 404 not found response
    if not todo:
        raise HTTPException(status_code=404, detail=f"todo item with id {id} not found")

    return todo

..only thing is, this is not correct. When we developed schemas.ToDo, we specifically excluded the id attribute because we intended it to be used in the create_todo() path operation function, where the user doesn’t know the id ahead of time. So, it actually makes more sense to create two separate schemas: one with an id attribute and one without it. We’ll update schemas.py like this

from pydantic import BaseModel

# Create ToDo Schema (Pydantic Model)
class ToDoCreate(BaseModel):
    task: str
    
# Complete ToDo Schema (Pydantic Model)
class ToDo(BaseModel):
    id: int
    task: str

And in main.py we’ll change create_todo(todo: schemas.ToDo) to create_todo(todo: schemas.ToDoCreate).

Let’s also improve the create() endpoint to return the created todo object instead of a string with its id. There’s a trick to make this work..

@app.post("/todo", response_model=schemas.ToDo, status_code=status.HTTP_201_CREATED)
def create_todo(todo: schemas.ToDoCreate):

    # create a new database session
    session = Session(bind=engine, expire_on_commit=False)

    # create an instance of the ToDo database model
    tododb = models.ToDo(task = todo.task)

    # add it to the session and commit it
    session.add(tododb)
    session.commit()
    session.refresh(tododb)

    # close the session
    session.close()

    # return the todo object
    return tododb

The trick is to call session.refresh(tododb) after session.commit(). If you don’t refresh the session, the tododb object doesn’t get updated with the database-generated id.

Cleaning up the read_todo_list() path operation is a bit trickier. Here, the expected response is a list of JSON objects. So, we add from typing import List and then modify the decorator like @app.get("/todo", response_model = List[schemas.ToDo]). However, we need to do one extra (not intuitive) thing; we need to set orm_mode = True in the ToDo schema.

schemas.py

# Complete ToDo Schema (Pydantic Model)
class ToDo(BaseModel):
    id: int
    task: str

    class Config:
        orm_mode = True

main.py

from typing import List
# <other imports not shown here>

@app.get("/todo", response_model = List[schemas.ToDo])
def read_todo_list():
    # create a new database session
    session = Session(bind=engine, expire_on_commit=False)

    # get all todo items
    todo_list = session.query(models.ToDo).all()

    # close the session
    session.close()

    return todo_list

The database session

There’s a lot of repeated code devoted to creating and closing the database session. There are a few things we can do to prevent that. Firstly, we can use sqlalchemy’s sessionmaker factory. With it, we can specify the session parameters we want once instead of making multiple calls like session = Session(bind=engine, expire_on_commit=False). I’ll implement this by adding the following to database.py

from sqlalchemy.orm import sessionmaker

# Create SessionLocal class from sessionmaker factory
SessionLocal = sessionmaker(bind=engine, expire_on_commit=False)

and then in main.py I can change every call like session = Session(bind=engine, expire_on_commit=False) to session = SessionLocal(). (Note also that I can remove from sqlalchemy.orm import Session but I need to add from database import SessionLocal).

This is an improvement, but we still have a lot of repeated code. Instead of creating a session inside each path operation function, we can insert a session parameter for each path operation. Furthermore, each of these session parameters can have a default value that references a get_session() function.

# Helper function to get database session
def get_session():
    session = SessionLocal()
    try:
        yield session
    finally:
        session.close()

then we can update our create_todo() function to

@app.post("/todo", response_model=schemas.ToDo, status_code=status.HTTP_201_CREATED)
def create_todo(todo: schemas.ToDoCreate, session: Session = Depends(get_session)):

    # create an instance of the ToDo database model
    tododb = models.ToDo(task = todo.task)

    # add it to the session and commit it
    session.add(tododb)
    session.commit()
    session.refresh(tododb)

    # return the todo object
    return tododb

Notes

I placed the get_session() helper function in main.py, right after app = FastAPI().
You’ll need to import Depends from fastapi and Session from sqlalchemy.orm
With this architecture, we don’t need to worry about closing the session inside the path operation function. The session is closed automatically once get_session() finishes execution.

More details about this in the FastAPI docs here.

Closing Thoughts

First of all, check out the final code for this project on github.

While I’m thrilled to get my todooo app working, it wasn’t exactly easy and there’s still a lot to do (heh). Next thing on my radar is deployment and understanding how to manage a production database in conjunction with a local database. Questions on my mind are things like How do I make my app always point to the correct database? and What happens if the structure of one of my database tables changes, making my Python models out of sync?

Contents

Building A Simple CRUD Application With FastAPI

Intro

Setup

Designing The App

Dummy App

The Database

SqlAlchemy Engine

SqlAlchemy Declarative

Database Initialization

Table Creation

Database Code Refactor

CRUD Operations

Create

Read

Update

Delete

Refactoring

Database Models vs Pydantic Models

Declaring Response Models

The database session

Closing Thoughts