Get records from the database with Python and SQLAlchemy

Views: 9
0 0
Read Time:6 Minute, 39 Second

This article explains how to get records from the database with Python and SQLAlchemy using the FastAPI framework for exposing a GET endpoint step by step. If you already don’t know, the GET is the R of CRUD that means (guess a bit) READ and is used for retrieving one or more records from the database.

In the previous articles we went through the initial setup of the CRUD project by adding stubs APIs, created the function for handling the POST and published the code to the gitlab repository.

Other articles of the same series

Project Setup: Project setup article

POST: Create a POST API endpoint

PUT: Create a PUT API endpoint

DELETE: Create a DELETE API Endpoint

What we are going to do in this part is to create the GET endpoint so that can actually write records to the database (which was created here).

This article will be way shorter because we can leverage entirely the ORM

We left our code here

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 / 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}"
        }

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

We are going to change the 2 GETs and in particular the GET All and the GET single

The code below is the one responsible for querying all the car records we already have in our table. We can easily achieve that by only adding one line of code

race_car_records = DBSession.query(RaceCarORM).filter().all()

everything else is just a side dish and the code is self explanatory

@app.get("/race-cars")
def get_all_cars(
    request: Request,
    response: Response,
):
    try:
        race_car_records = DBSession.query(RaceCarORM).filter().all()
        return {
            "entries": race_car_records,
            "total": len(race_car_records)
        }
    except Exception as e:
        response.status_code = status.HTTP_400_BAD_REQUEST
        return {
            "entries": [],
            "total":0, 
            "error": e,
            "error_details": e.orig.args if hasattr(e, 'orig') else f"{e}"
        }

The GET single instead is like below. The only difference is that in the filter function we are specifying the criteria that in our case is RaceCarORM.id must be equal to the car_id (the one passed by the request itself)

@app.get("/race-cars/{car_id}")
def get_car(
    car_id: int,
    request: Request,
    response: Response,
):
    try:
        race_car_record = DBSession.query(RaceCarORM).filter(RaceCarORM.id == car_id).first()
        if race_car_record:
            return {
                "entries": race_car_record,
            }
        else:
            return {
                "entries": [],
                "message": f"No entries found for id: {car_id}"
            }
    except Exception as e:
        response.status_code = status.HTTP_400_BAD_REQUEST
        return {
            "entries": [],
            "id_sent": car_id, 
            "total": 0,
            "error": e,
            "error_details": e.orig.args if hasattr(e, 'orig') else f"{e}"
        }

The full code should then looks like this 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-3",
    description="keep-4-yourself-example-03",
)

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

@app.get("/race-cars")
def get_all_cars(
    request: Request,
    response: Response,
):
    try:
        race_car_records = DBSession.query(RaceCarORM).filter().all()
        return {
            "entries": race_car_records,
            "total": len(race_car_records)
        }
    except Exception as e:
        response.status_code = status.HTTP_400_BAD_REQUEST
        return {
            "entries": [],
            "total":0, 
            "error": e,
            "error_details": e.orig.args if hasattr(e, 'orig') else f"{e}"
        }

@app.get("/race-cars/{car_id}")
def get_car(
    car_id: int,
    request: Request,
    response: Response,
):
    try:
        race_car_record = DBSession.query(RaceCarORM).filter(RaceCarORM.id == car_id).first()
        if race_car_record:
            return {
                "entries": race_car_record,
            }
        else:
            return {
                "entries": [],
                "message": f"No entries found for id: {car_id}"
            }
    except Exception as e:
        response.status_code = status.HTTP_400_BAD_REQUEST
        return {
            "entries": [],
            "id_sent": car_id, 
            "total": 0,
            "error": e,
            "error_details": e.orig.args if hasattr(e, 'orig') else f"{e}"
        }

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

# 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}"
        }

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

Super easy!

You can find the code used in this article here

https://github.com/keep4yourself/crud-records-with-fast-api-p1/tree/part-3

Hope you enjoyed this article and in the meantime if you can, please share and help us grow.

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

Happy
Happy
0 %
Sad
Sad
0 %
Excited
Excited
0 %
Sleepy
Sleepy
0 %
Angry
Angry
0 %
Surprise
Surprise
0 %
Previous post How to compress a Linux Directory
Next post Convert poetry.lock file into a requirements.txt and download the packages
0 0 votes
Article Rating
Subscribe
Notify of
guest
0 Comments
Inline Feedbacks
View all comments
0
Would love your thoughts, please comment.x
()
x