Create a POST API endpoint using Python, FastAPI and SQLAlchemy

by d3 · August 7, 2021

Post Views: 1,117

This article explains how to create a POST API endpoint using Python, FastAPI and SQLAlchemy step by step. If you already don’t know, the POST is the C of CRUD that means (guess a bit) Create. In the previous article (https://keepforyourself.com/coding/how-to-crud-records-with-fastapi/) we went through the initial setup of the CRUD project by adding stubs APIs and published the initial code to the gitlab repository https://github.com/keep4yourself/crud-records-with-fast-api-p1. What we are going to do in this part is to create the POST endpoint so that can actually write records to the database (which was created here).

Other articles of the same series

Project Setup: Project setup article

GET: Create a GET API endpoint

PUT: Create a PUT API endpoint

DELETE: Create a DELETE API Endpoint

This article will have the below sections:

Import the libraries and connect to the database
Create the database model for the race_car table
Change the post request so that the record can be written
End result
Verify the working b ehaviour
Gitlab-repository-link

Please note that this code is not production ready but is enough for showcase how immediate the implementation of a POST endpoint can be just by using this nice library.

Let’s clone the repository first

git clone https://github.com/keep4yourself/crud-records-with-fast-api-p1.git
# we can switch to where we left
git checkout part-1

We should have a main.py that is similar to the one below and what we should do at this stage is to

from typing import Optional
from pydantic import BaseModel
from fastapi import FastAPI

class RaceCar(BaseModel):
    id: Optional[int] = None
    car_number: int
    driver_name: str
    team_name: str

app = FastAPI(
    title="Example-02-CRUD-part-1",
    description="keep-4-yourself-example-02",
)

@app.get("/")
def say_hello():
    return {"hello": "world "}

@app.get("/race-cars")
def get_all_cars():
    return {"cars": ["all"]}

@app.get("/race-cars/{car_id}")
def get_car(
    car_id: int
):
    return {"car": [f"returning details for {car_id}"]}

@app.put("/race-cars/{car_id}")
def edit_car(
    car_id: int
):
    return {"car": [f"editing details for {car_id}"]}

@app.post("/race-cars/")
def create_car(
    race_car: RaceCar
):
    return {"car": race_car}

@app.delete("/race-cars/{car_id}")
def delete_car(
    car_id: int
):
    return {"car": [f"delete car {car_id}"]}

Import the libraries and connect to the database

For achieving this we should import the sqlalchemy related libraries

# declarative_base class, Column, Integer and String
# will all be used for the race_car table model
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine, Column, Integer, String
# Session will be used together wiith create_engine 
# for the connection session
from sqlalchemy.orm import Session

# my database is on the same machine 
# you should change the localhost with the IP of 
# your database
DB_HOST = "localhost" 
# the database we created in the previous article
# https://keepforyourself.com/databases/mysql/how-to-install-mysql-on-your-linux-system/
DATABASE = "playground"

engine = create_engine(f"mysql+pymysql://root:pass123@{DB_HOST}/{DATABASE}")
DBSession = Session(engine)

Create the database model for the race_car table

This will be our “model” or record representation. Each __tablename__ record entity will have the structure of below

class RaceCarORM(DB_BASE_ORM):
    __tablename__ = "race_cars"
    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    car_number = Column(Integer, index=False)
    driver_name = Column(String, index=False)
    team_name = Column(String, index=False)

Change the post request so that the record can be written

The code below is pretty immediate to read, but just in case what we are doing here is to

get the request object and convert to an instance of the model (RaceCarORM type)
create the record (add + commit)
return the result

In case the car_number is duplicate, an exception will be raised and the error returned together with a 400 bad request.

# remove the ending / that will cause a redirect for the post request
# issue hit by the green guy :D
@app.post("/race-cars")
def create_car(
    request: Request,
    response: Response,
    race_car: RaceCar,
):
    try:
        DBSession.begin()
        race_car_record = RaceCarORM(**dict(race_car))
        DBSession.add(race_car_record)
        DBSession.commit()
        race_car.id = race_car_record.id
        return race_car
    except Exception as e:
        DBSession.rollback()
        response.status_code = status.HTTP_400_BAD_REQUEST
        return {
            "error": e,
            "error_details": e.orig.args if hasattr(e, 'orig') else f"{e}"
        }

End result

The ending result should be a main.py file like the one below

from typing import Optional
from pydantic import BaseModel
from fastapi import FastAPI, Request, Response, status
# declarative_base class, Column, Integer and String
# will all be used for the race_car table model
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine, Column, Integer, String
# Session will be used together wiith create_engine 
# for the connection session
from sqlalchemy.orm import Session

# my database is on the same machine 
# you should change the localhost with the IP of 
# your database
DB_HOST = "localhost" 
# the database we created in the previous article
# https://keepforyourself.com/databases/mysql/how-to-install-mysql-on-your-linux-system/
DATABASE = "playground"

engine = create_engine(f"mysql+pymysql://root:pass123@{DB_HOST}/{DATABASE}")
DBSession = Session(engine)

DB_BASE_ORM = declarative_base()

class RaceCar(BaseModel):
    id: Optional[int] = None
    car_number: int
    driver_name: str
    team_name: str

class RaceCarORM(DB_BASE_ORM):
    __tablename__ = "race_cars"
    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    car_number = Column(Integer, index=False)
    driver_name = Column(String, index=False)
    team_name = Column(String, index=False)

app = FastAPI(
    title="Example-02-CRUD-part-2",
    description="keep-4-yourself-example-03",
)

@app.get("/")
def say_hello():
    return {"hello": "world"}

@app.get("/race-cars")
def get_all_cars():
    return {"cars": ["all"]}

@app.get("/race-cars/{car_id}")
def get_car(
    car_id: int
):
    return {"car": [f"returning details for {car_id}"]}

@app.put("/race-cars/{car_id}")
def edit_car(
    car_id: int
):
    return {"car": [f"editing details for {car_id}"]}

# remove the ending / from the decorator because it will cause a 
# redirect for the post request (issue hit by the green guy :D)
@app.post("/race-cars")
def create_car(
    request: Request,
    response: Response,
    race_car: RaceCar,
):
    try:
        DBSession.begin()
        race_car_record = RaceCarORM(**dict(race_car))
        DBSession.add(race_car_record)
        DBSession.commit()
        race_car.id = race_car_record.id
        return race_car
    except Exception as e:
        DBSession.rollback()
        response.status_code = status.HTTP_400_BAD_REQUEST
        return {
            "error": e,
            "error_details": e.orig.args if hasattr(e, 'orig') else f"{e}"
        }

@app.delete("/race-cars/{car_id}")
def delete_car(
    car_id: int
):
    return {"car": [f"delete car {car_id}"]}

The above code should be enough for accepting the POST request and write in the database.

Verify the working behaviour

# spin up the application
uvicorn main:app --host 0.0.0.0 --port 8080 --reload

Send a curl request

curl -X POST "http://192.168.1.11:8080/race-cars" --data '{"car_number": 33, "driver_name": "Max Verstappen", "team_name": "Red Bull Racing F1"}' -H "Content-Type: application/json"

The above expression should have as end result the car record added in the database, and for verifying that we can run this command

mysql> select * from race_cars;
+----+------------+----------------+--------------------+
| id | car_number | driver_name    | team_name          |
+----+------------+----------------+--------------------+
| 61 |         33 | Max Verstappen | Red Bull Racing F1 |
+----+------------+----------------+--------------------+

Gitlab repository link

Gitlab repo

I hope you could find this helpful, in the next parts we will try to give a little bit of structure to the code and go through the remaining methods, in the meantime if you can, please share.

If you are looking for a nice and well organized book that explains the core fundamentals around FastAPI we recommend this book:

Create a POST API endpoint using Python, FastAPI and SQLAlchemy

Other articles of the same series

Import the libraries and connect to the database

Create the database model for the race_car table

Change the post request so that the record can be written

End result

Verify the working behaviour

Gitlab repository link

Like this:

You may also like...

Other articles

Create a POST API endpoint using Python, FastAPI and SQLAlchemy

Other articles of the same series

Import the libraries and connect to the database

Create the database model for the race_car table

Change the post request so that the record can be written

End result

Verify the working behaviour

Gitlab repository link

Share this:

Like this:

You may also like...

Python Decorators, Superpowers for code

Switch python version using pyenv

Learning PHP – Lesson 1 – print and echo

Other articles