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

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.

Josh's Blog; Data.Code.Cloud