Building a REST API with Python FastAPI and PostgreSQL

Introduction

In this tutorial, we will guide you through the process of creating a REST API using Python FastAPI and PostgreSQL. We will connect to a PostgreSQL database named coffeecms and create two endpoints:

• api/GetAllUser: This endpoint will display a list of all the records in the users table.
• api/UpdateUser: This endpoint will update the fullname and email fields in the users table for a given user_id.

Prerequisites

Before you begin, you will need to have the following installed on your system:

• Python 3.6 or later
• PostgreSQL
• FastAPI

Creating the PostgreSQL Database

1. Create a PostgreSQL database named coffeecms. You can do this using the following command:

psql -U postgres -c "CREATE DATABASE coffeecms"

2. Create a table named users in the coffeecms database:

CREATE TABLE users (
  user_id SERIAL PRIMARY KEY,
  fullname VARCHAR(255) NOT NULL,
  email VARCHAR(255) NOT NULL
);

3. Insert some data into the users table:

INSERT INTO users (fullname, email) VALUES
  ('John Doe', '[email protected]'),
  ('Jane Doe', '[email protected]'),
  ('Peter Jones', '[email protected]');

Creating the FastAPI Application

1. Create a new directory for your FastAPI application.
2. Create a file named main.py in the directory you created in step 1.
3. Add the following code to the main.py file:

from fastapi import FastAPI, Depends
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from pydantic import BaseModel

from database import get_db

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

class User(BaseModel):
    user_id: int
    fullname: str
    email: str

class UserIn(BaseModel):
    fullname: str
    email: str

class UserUpdate(BaseModel):
    fullname: str
    email: str

engine = create_engine("postgresql://user:password@localhost/coffeecms")

def get_db(db=Depends(oauth2_scheme)):
    db = Session(local_session=db)
    try:
        yield db
    finally:
        db.close()

@app.get("/api/GetAllUser", tags=["Users"])
async def get_all_users(db: Session = Depends(get_db)):
    users = db.query(User).all()
    return users

@app.post("/api/CreateUser", tags=["Users"])
async def create_user(user: UserIn, db: Session = Depends(get_db)):
    new_user = User(fullname=user.fullname, email=user.email)
    db.add(new_user)
    db.commit()
    return new_user

@app.put("/api/UpdateUser/{user_id}", tags=["Users"])
async def update_user(user_id: int, user: UserUpdate, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.user_id == user_id).first()
    if db_user is None:
        return {"message": "User not found"}
    db_user.fullname = user.fullname
    db_user.email = user.email
    db.commit()
    return {"message": "User updated successfully"}

4. Create a file named database.py in the same directory as main.py.
5. Add the following code to the database.py file:

from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

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

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

Running the Application

1. Open a terminal window and navigate to the directory where you saved the main.py file.
2. Run the following command to start the FastAPI application: uvic