Introduction
This article will continue our journey and connect a FastAPI project to a SQL Server database. We'll walk through a full CRUD (Create, Read, Update, Delete) example of saving data through our FastAPI project to a database. We'll work with a healthcare example and manage a Patient table. Of course, we aren't reviewing database design, so we are working with a single table for now. You should already be familiar with databases and have sufficient access required to create tables and read and insert from a database.
Code for this example can be found here - API SQL Server DB CRUD Repo. The README provides startup steps.
For this article, I'm working with a local SQL Server 2017 Database on my mac. Before continuing, if you don't have a SQL Server playground follow my article to set up SQL Server on your mac. This will give you a fully functioning SQL Server Database running in a Docker container on your local machine.
We'll also be using SQL Alchemy as the database ORM. SQL Alchemy is a well-known ORM project to connect SQL databases.
Toolkit
- FastAPI - SQL Databases
- SQL Alchemy ORM 1.4
- PyMsSql
- Pydantic Schemas/ Models
- SQL Server 2017
- Docker Desktop
Database Setup
At this point, you should already have a SQL Server Db to work with. In the project repository, run the Db_Scripts/SEED_Script.sql
file to create the Patient Table and add a single record to it. Running a SELECT
statement shows the single record in the table. My database is named MyTestDb
, change any of the SQL Statements to make your database name when necessary.
FastAPI Db Connection
Using SQL Alchemy, we need a connection.py
file to setup a database connection. We'll create this connection as a dependency to use in our API routers.
First things first, when using SQL Alchemy an engine
is required to make connections to databases.
Once the engine
is created, we use SQL Alchemy's sessionmaker
to create a session
. The session
serves as a collection of queries/ statements that we call to execute in the database. The session
will hold these events until the session
is instructed to either commit
or rollback
the events.
When making a database call in our API, we call a method get_db
to execute the following try/catch
statement to start a database session
. Once we execute our queries/ instructions in this session
, the finally
statement will also close the session
when it is complete.
# taken from utils/connection.py
...
# Create engine
engine = create_engine(f'mssql+pymssql://{settings.DB_UID}:{settings.DB_PWD}@{settings.DB_SERVER}:{settings.DB_PORT}/{settings.DB_NAME}')
# Create Session
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
...
try:
db = SessionLocal()
yield db
finally:
db.close()
To use these database sessions, we'll import our utils
module into the app.py
file.
# app.py
import utils
In our API Endpoints, we insert the database connection as a dependency in the method parameter. We now have the db
instance to call database instructions.
# app.py
# CRUD Routes
# GET ALL
# Response will be a LIST of schema.Patient
# The Schema.Patient List instance will be mapped from the model.Patient ORM instance from Sql Alchemy
@app.get("/Patient", response_model=List[schema.Patient])
async def GetAll(db: Session = Depends(utils.get_db)):
# get the model.patient with the given id
# using sql alchemy orm, we're querying the Patient table
query = db.query(model.Patient)
patients = query.all()
return patients
Now have connectivity to our database, we need to discuss creating Models to interact with our database.
Models Vs Schemas
ORM Models SQL Alchemy
We're utilizing two different representations of our Patient table in the API, an ORM Model (SQL Alchemy) and a Schema Model (Pydantic).
The ORM Model is a SQL Alchemy representation of the Patient Table. This ORM model knows the database table that queries are going to be executed on by the __tablename__= 'Patient'
statement. The ORM model has a direct representation of the database table fields. This assists with validation when trying to save or edit Patient
fields.
ORM Models are made from the Delcarative Base SQL Alchemy Class in the utils/connection.py
file. The code to create an ORM model is shown below.
# Patient/model.py
from sqlalchemy import Column, Integer, String, DateTime
from utils.connection import Base
# Base is coming from the SQL Alchemy Base class created in utils/connection.py
class Patient(Base):
__tablename__ = 'Patient'
ID = Column(Integer, primary_key=True)
BirthDate = Column(DateTime)
FirstName = Column(String(50))
LastName = Column(String(50))
AccountNumber = Column(String(25))
Department = Column(String(20))
Room = Column(String(1))
Schema Models (Pydantic)
The Schema models on the other hand are Pydantic models that can be used in the API to also:
- validate data
- identify endpoint
response
types - be used in the Swagger or Redoc Documentation pages
The Schemas are created from the Pydantic BaseModel
class, unlike the ORM Models.
Our Patient table has two schemas. The PatientCreate
schema is used as the body request when inserting a new record in our post
endpoint. The ID
field is not part of this schema so the API can only validate only the required fields.
The second schema is the same representation of the Patient ORM model. Type schema has the orm_mode
value set to true
in order to support mapping this schema to the Patient ORM Model when returning model types. This is specifically used when we get a Patient or Patient list in our endpoints. We're using the ORM SQL Alchemy model to return the records from the database. The data is then validated and returned in the API as a Pydantic model. This is shown in the endpoint decorater:
@app.get("/Patient", response_model=List[schema. Patient])
from datetime import datetime
from pydantic import BaseModel
# Create Patient Schema (Pydantic Model)
class PatientCreate(BaseModel):
BirthDate: datetime
FirstName: str
LastName: str
AccountNumber: str
Department: str
Room: str
class Patient(BaseModel):
ID: int
BirthDate: datetime
FirstName: str
LastName: str
AccountNumber: str
Department: str
Room: str
# ORM Mode is used to support models that map to ORM objects, in this case model.Patient (sqlAlchemy)
class Config:
orm_mode = True
Schemas can be used in different scenarios when validating or specifying only specific elements of a class object.
CRUD
Get All - READ
This endpoint will return all Patient records. No request parameters are included in this endpoint. Fast API identifies the get method by the decorator @app.get
command.
The route is /Patient
and the response is a list of the schema.Patient
model.
# app.py
# GET ALL
# Response will be a LIST of schema.Patient
# The Schema.Patient List instance will be mapped from the model.Patient ORM instance from Sql Alchemy
@app.get("/Patient", response_model=List[schema.Patient])
async def GetAll(db: Session = Depends(utils.get_db)):
# get the model.patient with the given id
# using sql alchemy orm, we're querying the Patient table
query = db.query(model.Patient)
patients = query.all()
return patients
Get Single - READ
This endpoint will return a single Patient record. A single ID
parameter is included in this endpoint.
The route is /Patient/{ID}
and the response is a single record of the schema.Patient
model.
If a Patient with the provided ID
is not found a 404, not found
response is returned.
# app.py
# GET Single
# Response will be a single schema.Patient
# The Schema.Patient instance will be mapped from the model.Patient ORM instance from Sql Alchemy
@app.get("/Patient/{ID}")
async def GetSingle(ID: int, db: Session = Depends(utils.get_db)):
# get the patient with the given Patient ID
query = db.query(model.Patient).filter(model.Patient.ID == ID)
patient = query.one_or_none()
if not patient:
raise HTTPException(status_code=404, detail=f"Patient with ID {ID} not found")
return patient
Post - Write
This endpoint will insert a single Patient record. A request body in JSON format is included in this endpoint. The request body will need to match the PatientCreate
schema specified in the method parameter.
The route is /Patient
and the response is a single record of the Patient model that was just inserted into the database.
After the commit
command a refresh
command is called. This returns the Patient model with the newly generated Patient ID. The response will now include the Patient ID which is the primary key value of the Patient.
# app.py
# POST
# Response will be a single schema.Patient after creation in the DB
# The Schema.Patient instance will be mapped from the model.Patient ORM instance from Sql Alchemy
@app.post("/Patient", response_model=schema.Patient)
async def Post(patient: schema.PatientCreate, db: Session = Depends(utils.get_db)):
# create an instance of the model.Patient ORM model from the schema.Patient instance from the request body
new_patient = model.Patient(BirthDate = patient.BirthDate
, FirstName = patient.FirstName
, LastName = patient.LastName
, AccountNumber = patient.AccountNumber
, Department = patient.Department
, Room = patient.Room)
# add it to the session and commit it
db.add(new_patient)
db.commit()
# update the patient instance to get the newly created Id
db.refresh(new_patient)
# return the patient
return new_patient
Put - Write
This endpoint will update a single Patient record. A request body in JSON format is included in this endpoint. The request body will need to match the PatientCreate
schema specified in the method parameter.
The route is /Patient/{ID}
and the response is a single record of the Patient model that was just updated into the database.
A query for the Patient ID
is the first step to getting the existing Patient. Once found, fields are updated based on the request body schema.PatientCreate
model. After the commit
statement a refresh
command is called. This returns the Patient model with the newly updated Patient fields.
If a Patient with the provided ID
is not found a 404, not found
response is returned.
# app.py
# PUT
# Response will be a single schema.Patient after creation in the DB
# The Schema.Patient instance will be mapped from the model.Patient ORM instance from Sql Alchemy
@app.put("/Patient/{ID}", response_model=schema.Patient)
async def Put(ID: int, patientUpdate: schema.Patient, db: Session = Depends(utils.get_db)):
# get the model.Patient with the given id
patient = db.get(model.Patient, ID)
# update patient with the patient from request body (if patient with the given id was found)
if patient:
patient.BirthDate = patientUpdate.BirthDate
patient.FirstName = patientUpdate.FirstName
patient.LastName = patientUpdate.LastName
patient.AccountNumber = patientUpdate.AccountNumber
patient.Department = patientUpdate.Department
patient.Room = patientUpdate.Room
db.commit()
db.refresh(patient)
# check if patient with given id exists. If not, raise exception and return 404 not found response
if not patient:
raise HTTPException(status_code=404, detail=f"Patient with ID {ID} not found")
return patient
Delete - Write
This endpoint will delete a single Patient record. A single ID
parameter is included in this endpoint.
The route is /Patient/{ID}
and the response a 204, No Content
upon completion with no errors.
# app.py
# DELETE
@app.delete("/Patient/{ID}", status_code=status.HTTP_204_NO_CONTENT)
async def Delete(ID: int, db: Session = Depends(utils.get_db)):
# get the model.Patient with the given id
patient = db.get(model.Patient, ID)
# check if patient with given id exists and call delete
if patient:
db.delete(patient)
db.commit()
End
This article reviewed the steps to set up a SQL Alchemy connection to a SQL Server database.
This simple CRUD API project shows the steps to read and write data from a SQL Server database table for persistent storage.