FastAPI app with PostgreSQL Tutorial

In this FastAPI with PostgreSQL Tutorial, we’ll delve into FastAPI, a powerful python web framework and SQLAlchemy, a SQL toolkit that enhances the flexibility of SQL.

Introduction

In this project, we will create a QUIZ Game, we will setup FastAPI with PostgreSQL. PostgreSQL is one of the most popular relational databases system management.

First thing, create a virtual environment and activate it:

python -m venv myenv
source myenv/bin/activate

FastAPI, SQLAlchemy, uvicorn, pydantic installation

Install all the dependencies needed for this application:

pip install fastapi sqlalchemy psycopg2-binary uvicorn

Start FastAPI Application

Create a main.py file:

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from typing import List, Annotated

app = FastAPI()

Let’s create two pydantic base models for Choice and Question:

class ChoiceBase(BaseModel):
    choice_text: str
    is_correct: bool

class QuestionBase(BaseModel):
    question_text: str
    choices: List[ChoiceBase]

Create PostgreSQL connection

Before creating our API endpoints, we will create our data models models.py for our Postgres database and create the database.py that will connect it all together our FastAPI application with our Postgres database. Create database.py:

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

# URL string for the Postgres database
# In this example, the database name is 'quizApp'
URL_DATABASE = 'postgresql://USERNAME:PASSWD@localhost:5432/quizApp'

engine = create_engine(URL_DATABASE)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

Create PostgreSQL Tables

Create models.py:

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from database import Base

class Questions(Base):
    __tablename__ = 'questions'

    id = Column(Integer, primary_key=True, index=True)
    question_text = Column(String, index=True)

class Choices(Base):
    __tablename__ = 'choices'

    id = Column(Integer, primary_key=True, index=True)
    choice_text = Column(String, index=True)
    is_correct = Column(Boolean, default=False)
    question_id = Column(Integer, ForeignKey("questions.id"))

Now go back to our main.py file and import the following:

import models
from database import engine, SessionLocal
from sqlalchemy.orm import Session

then under app = FastAPI(), add the following:

# Create all the tables and columns in PostgreSQL
models.Base.metadata.create_all(bind=engine)

Connect FastAPI, PostgreSQL with SQLAlchemy

Now after ChoiceBase and QuestionBase classes, create a connection to the database in a get_db() method:

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Then create this important annotation after the get_db() method:

# This will be used later for a dependency injection
db_dependency = Annotated[Session, Depends(get_db)]

FastAPI Add Question Endpoint

Now let’s create our API endpoints. We will create the add question endpoint, at the end of main.py add the following:

# We passing in a data validation to validate the body of the API request
# and passing the database connection
@app.post('/questions/')
def create_questions(question: QuestionBase, db: db_dependency):
    # Here we use SQL Alchemy to write an ORM statement that will link with the database
    db_question = models.Questions(question_text=question.question_text)
    db.add(db_question)
    db.commit()
    db.refresh(db_question)
    for choice in question.choices:
        db_choice = models.Choices(choice_text=choice.choice_text, is_correct=choice.is_correct, question_id=db_question.id)
        db.add(db_choice)
    db.commit()

Let’s run our application:

uvicorn main:app --reload

Test your API endpoint in FastAPI Swagger with the following data:

{
  "question_text": "What is the best Python Framework",
  "choices": [
    {
      "choice_text": "FastAPI",
      "is_correct": true
    },
    {
      "choice_text": "Flask",
      "is_correct": false
    },
    {
      "choice_text": "Django",
      "is_correct": false
    }
  ]
}

Other FastAPI Endpoints

Let’s add a Get request method to be able to fetch a question:

@app.get('/questions/{question_id}')
def read_questions(question_id: int, db: db_dependency):
    result = db.query(models.Questions).filter(models.Questions.id == question_id).first()
    if not result:
        raise HTTPException(status_code=404, detail='Question is not found!')
    return result

Let’s add a Get request method to be able to fetch a question choices:

@app.get('/choices/{question_id}')
def read_choices(question_id: int, db: db_dependency):
    result = db.query(models.Choices).filter(models.Choices.question_id == question_id).all()
    if not result:
        raise HTTPException(status_code=404, detail='Choices is not found!')
    return result

Test all API endpoints

By Wahid Hamdi