CRUD operations with FastApi

FastApi is a Python framework aimed at API development. In order to achieve great performances, it leverages asynchronous code execution powered by coroutines, a language feature available since Python release 3.6. In this article we will create a simple CRUD API (Create, Read, Update, Delete) using the tools provided by FastApi.

The specific example deals with a simple memo api, in which each memo is described by a title and a description.

So, let's begin!

1. Installation of dependencies

For our project we will need some dependencies:

• FastApi: to create the core API logic
• SQLAlchemy: in order to integrate a database for persistence
• Uvicorn: the server implementation that will execute our API

Let's start creating a directory for our project and a Python virtual environment to install dependencies in:

# Directory creation
$ mkdir fastapi-items
$ cd fastapi-items
# Initialization of the virtual environment
$ python3 -m venv env
# Activation of the environment
$ source env/bin/activate

Now, install dependencies

(env) $ pip install fastapi "uvicorn[standard]" sqlalchemy

2. Code layout

Please, create the following tree of files in folder in order to follow our steps

.
└── env
├── items_app
    ├── __init__.py
    ├── crud.py
    ├── database.py
    ├── main.py
    ├── models.py
    └── schemas.py

This is the purpose of each file

• init.py: makes the items_app directory a python module, so that we can import items from one file to another
• crud.py: will contain the implementation of the crud operations against the database. It is important to keep this separated from the API for testing purposes.
• database.py: will contain the necessary logic to interact with the database
• main.py: application entry point, will contain the definition of the API
• models.py: this will contain the definition of the tables that will be stored in the database
• schemas.py: this will contain the definition of the shapes of requests and responses that our api will deal with

3. Database connection

We will use a SQLite database for this example, but the approach is quite general and can be easily adapted to any DBMS.

# database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

The elements defined in this file represent

• engine: manager of connections to the database
• SessionLocal: class whose instances are the effective communication channels with the DBMS
• Base: base class for all the models used by the ORM (Object Relational Mapping)

4. Data model definition

For the sake of the example, just one data model is needed with the following fields

• id: unique identifier of each instance, integer autoincrementing
• title: title of the object (string type)
• description: extended description (free text)

By using the primitives provided by SQLAlchemy, the definition of our data model is the following:

# models.py
from sqlalchemy import Column, Integer, String

from .database import Base


class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(String)

Beware: the Item class extends the Base class that we defined previously in the database.py file. This in necessary in order to allow the ORM to work properly and to let SQLAlchemy create the correct tables in the database at boot time.

5. Definition of the representations of the data

The intrinsic nature of the different operations exposed by the API require that not all attributes of an object are present in every context. For instance, the id attribute, which identifies an object resident in the database, is usually auto-generated when the object is sent to persistence, and so it cannot be generated by the client asking the server to create the object.

In our example, we are defining three representations (aka schemas), for three different usages:

• ItemCreate represents an Item that a client asks to be created
• ItemUpdate represents an Item that a client asks to be updated
• Item represents an Item when it is sent out from the api in response of some user input

# schemas.py
from typing import Optional

from pydantic import BaseModel


class ItemCreate(BaseMode):
    title: str
    description: Optional[str] = None


class ItemUpdate(BaseModel):
    title: str
    description: Optional[str] = None


class Item(ItemBase):
    id: int
    title: str
    description: Optional[str] = None

    class Config:
        orm_mode = True

The Item schema is configured to operate in orm_mode: this tells pydantic to operate on object with attributes and not on dicts with keys and values.

These "schemas" have a role similar to that of a serializer in django-rest-framework

6. Definition of CRUD operations against the database

We will now define the basic operations of a CRUD api, which are:

• list: allows to retrieve a list of the instances of a known class
• retrieve: allows to get a designated entity from a class, known its identifier
• create: crafts a new entity of a known class and persists it to the database
• update: changes attributes of a known entity, targeted by its identifier, and persists changes in the database
• delete: drops an entity from persistence, known its identifier

The implementation is designed to be easily used by the clients of our API: the parameters sento to each function are of typed with schemas (for instance ItemUpdate o ItemCreate), while the returned values are instances of the data model classes (Item item class from file models.py)

The first parameter of each function (db) represents the database connection, and it is necessary for the operation itself. All the other paramenters vary case by case. For instance the create operation requires data to craft a new Item.

# crud.py
from sqlalchemy.orm import Session

from .models import Item
from .schemas import ItemCreate, ItemUpdate
from typing import Union


def list_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(Item).offset(skip).limit(limit).all()


def get_item(db: Session, id: int):
    return db.query(Item).get(id)


def create_item(db: Session, data: ItemCreate):
    db_item = Item(**data.dict())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item


def drop_item(db: Session, item_id: int):
    db.query(Item).filter(Item.id == item_id).delete()
    db.commit()
    return None


def update_item(db: Session, item: Union[int, Item], data: ItemUpdate):
    if isinstance(item, int):
        item = get_item(db, item)
    if item is None:
      return None
    for key, value in data:
        setattr(item, key, value)
    db.commit()
    return item

7. API Implementation

The implementation of the API relies on all the building blocks we created so far:

• data models manage the interactions with the database
• schemas define the shapes of data entering and exiting the API
• the database connection allows concrete data access

This is the implementation of the CRUD operations on the Item model

# main.py
from typing import List, Optional
from fastapi import FastAPI, HTTPException
from fastapi.params import Depends
from sqlalchemy.orm import Session

from . import models, crud, schemas
from .database import engine, SessionLocal

# Auto creation of database tables
#   If tables already exist, this command does nothing. This allows to 
#   safely execute this command at any restart of the application.
#   For a better management of the database schema, it is recommended to
#   integrate specific tools, such as Alembic
models.Base.metadata.create_all(bind=engine)

# Application bootstrap
app = FastAPI()


# This function represents a dependency that can be injected in the endpoints of the API.
# Dependency injection is very smart, as it allows to declaratively require some service.
# This function models the database connection as a service, so that it can be required
# just when needed.
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


# LIST
# This endpoint returns a list of objects of type `Item` serialized using the `Item` schema that
# we defined in schemas.py. The objects exposed are instances of `models.Item` that are
# validated and serialized as of the definition of the schema `schemas.Item`
@app.get("/items", response_model=List[schemas.Item])
def items_action_list(limit: int = 100, offset: int = 0, db: Session = Depends(get_db)):
    items = crud.list_items(db, offset, limit)
    return items

# RETRIEVE
# This endpoint returns a specific `Item`, given the value of its `id` field, 
# which is passed as a path parameter in the URL. It can also return some 
# error condition in case the identifier does not correspond to any object
@app.get("/items/{item_id}", response_model=schemas.Item)
def items_action_retrieve(item_id: int, db: Session = Depends(get_db)):
    item = crud.get_item(db, item_id)
    if item is None:
        raise HTTPException(status_code=404)
    return item

# CREATE
# This endpoint creates a new `Item`. The necessary data is read from the request
# body, which is parsed and validated according to the ItemCreate schema defined beforehand
@app.post("/items", response_model=schemas.ItemCreate)
def item_action_create(data: schemas.ItemCreate, db: Session = Depends(get_db)):
    item = crud.create_item(db, data)
    return item

# UPDATE
# This endpoint allows to update an existing `Item`, identified by its primary key passed as a  
# path parameter in the url. The necessary data is read from the request
# body, which is parsed and validated according to the ItemUpdate schema defined beforehand
@app.put("/items/{item_id}", response_model=schemas.Item)
def items_action_retrieve(item_id: int, data: schemas.ItemUpdate,  db: Session = Depends(get_db)):
    item = crud.update_item(db, item_id, data)
    if item is None:
        raise HTTPException(status_code=404)
    return item


# DELETE
# This endpoint allows to delete an `Item`, identified by its primary key passed as a  
# path parameter in the url. It's worth observing that the status code of the response is
# HTTP 204 No Content, since the response body is empty
@app.delete("/items/{item_id}", status_code=204)
def items_action_retrieve(item_id: int,  db: Session = Depends(get_db)):
    crud.drop_item(db, item_id)
    return None

8. Execution of the code

In order to run our API it is enough to run this simple yet powerful command

(env) $ uvicorn items_app.main:app --reload --lifespan off

Now the API is listening for requests on port 8000, and it is possible to interact with it with any REST tools, such as Postman or Insomnia

References

Contents of the articles are inspired by the official FastApi tutorial. We really recommend reading the docs, as it is very complete and well organized.

Why FastApi

With respect to more traditional Python frameworks, such as Django or Flask, FastApi leverages more modern stardard interfaces (ASGI vs WSGI) and is really more powerful in benchmarks and effective in resource usage. Moreover, the base components of FastApi rely on the most modern language features, such as typings. This awards our code with great clarity and legibility, plus it improves the support by IDEs and simply error diagnostic.

---

The full source code that we developed in this article is available on our GitHub repository, and can be used as a starting point for the definition of any REST api.

The case we discussed today is deliberately simple, in order to build a solid conceptual foundation of the framework. This can be a valid starting point to learn advanced concepts, such as security elements or a more organized way to deal with data model evolution and update. If you liked this article or if you are interested in knowing more, let us now! You can find us on facebook and twitter.