Get records from the database with Python and SQLAlchemy
0
0 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
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
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!
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:
Share this content:
