Introduction

In this tutorial i will be teaching you how to build RESTful API using the python fastest framework know as FastAPI with postgres database, but you are free to choose any database of your choice.

In this tutorial I will be focusing on:

1. Todo CRUD API (Create, Read, Update, Delete)

2. Authentication using JWT

3. Authorization

FastAPI framework, high performance, easy to learn, fast to code, ready for production

Read more about FastAPI

Installation

Make sure you download and install python if you don't have it already install in your system.

Now let us install fastAPI, but before then create a folder where your fastAPI project will reside, (you can call it tutorials)

In side the tutorial directory create a folder called todo-api

To confirm if python and pip is install run the following commands

python --version

pip is a package manager for python, just like npm for nodejs

pip--version

Navigate into the todo-api and Run the following command

• To install fastapi

pip install fastapi

• uvicorn: uvicorna is a package manager for python that will be responsible for restarting our server after a change in our file. (uvicorn is just like nodemon in npm)

pip install uvicorn

• Install sqlalchemy aka ORM

pip install sqlalchemy

• Install jwt packages

pip install PyJWT 
pip install python-jose

• Install bcrypt for password hashing

pip install  bcrypt

• Install python-dotenv for .env

pip install python-dotenv

Building the todo api

Create a folder called app in the todo-api, the app folder is where all our code directory something like src in expressjs.

Create a file called main.py in the app directory i.e app/main.py and paste this code

from fastapi import FastAPI

from app import models
from app.database import engine
from app.routers import users_router
from app.routers import todos_router

models.Base.metadata.create_all(bind=engine)

app = FastAPI()
app.include_router(users_router.router)
app.include_router(todos_router.router)


@app.get("/")
async def root():
    return {"message": "Hello World"}

The main.py is the entry point of our application which contain the routers and our database connection

create a file models.py in the app directory i.e app/models.py and paste the code

from app.database import Base
from sqlalchemy import Column, String, Integer, ForeignKey
from sqlalchemy.orm import relationship


class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    password = Column(String)
    todos = relationship("Todo", back_populates="user")


class Todo(Base):
    __tablename__ = 'todos'

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, unique=True, index=True)
    user_id = Column(Integer, ForeignKey("users.id"))
    user = relationship("User", back_populates="todos")

The models.py is the database entity

Create a file called schema.py and paste the code

from fastapi import Query
from pydantic import BaseModel, Required, EmailStr, validator
from typing import Union


class User(BaseModel):
    email: Union[str, None] = Query(default=Required)
    password: Union[str, None] = Query(default=Required, min_length=4)


class Todo(BaseModel):
    title: Union[str, None] = Query(default=Required, min_length=2)


class Token(BaseModel):
    access_token: str
    token_type: str


class TokenData(BaseModel):
    email: Union[str, None] = None
    id: int

The schema.py file is the pydantic model something like interface in typescript and we use it for validation

Now, let connect to database. create a file called database.py and paste this code

import os
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from dotenv import load_dotenv

load_dotenv()
engine = create_engine(os.environ.get('SQLALCHEMY_DATABASE_URL'))
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()


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

Before we forget in your app folder create a file called pycache.py and init.py

Authentication

Now let create sign in and sign up functionalities.

Create a folder in the app directory called routers i.e app/routers, inside the folder create a file init.py to make it a module, also create a file called user_router.py

user_router.py file

from fastapi import Depends, APIRouter, Response
from sqlalchemy.orm import Session
from app import schema
from app.database import get_db
from app.repository import user_repository

router = APIRouter(prefix="/api/v1/users", tags=["users"], )


@router.post('/sign_up', status_code=201, tags=['users'])
def sign_up(request: schema.User, response: Response, db: Session = Depends(get_db)):
    return user_repository.sign_up(request, response, db)


@router.post('/sign_in', status_code=200, tags=['users'])
def login(request: schema.User, response: Response, db: Session = Depends(get_db)):
    return user_repository.login(request, response, db)


@router.get("/{user_id}", status_code=200)
def show(user_id: int, response: Response, db: Session = Depends(get_db)):
    return user_repository.show(user_id, response, db)

I have provide a standard code abstraction, so create another folder in the app directory call repository, this folder will be responsible for all our business logic (remember to create init.py file in the directory)

Inside the repository folder, create a file called user_repository.py and paste this code

from fastapi import Depends
from sqlalchemy.orm import Session
from starlette import status
from app import models, schema
from app.database import get_db
from app.helpers import password_helper
from app.helpers.jwt_helper import create_access_token


def sign_up(request: schema.User, response, db):
    email = db.query(models.User).filter(models.User.email == request.email).first()
    if email:
        response.status_code = status.HTTP_409_CONFLICT
        return {
            'message': "User already exist",
            'status_code': 409,
            'error': 'CONFLICT'
        }

    hash_password = password_helper.hash_password(request.password)
    new_user = models.User(email=request.email, password=hash_password)
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return {
        'message': "success",
        'status_code': 201,
        'status': 'Success',
        'data': {
            'id': new_user.id,
            'email': new_user.email
        }
    }


def login(request: schema.User, response, db):
    user: models.User = db.query(models.User).filter(models.User.email == request.email).first()
    if not user:
        response.status_code = status.HTTP_403_FORBIDDEN
        return {
            'message': "Invalid email and/or password",
            'status_code': 403,
            'error': 'FORBIDDEN'
        }

    if not (password_helper.verify_password(request.password, user.password)):
        response.status_code = status.HTTP_403_FORBIDDEN
        return {
            'message': "Invalid email and/or password",
            'status_code': 403,
            'error': 'FORBIDDEN'
        }
    #  Generate jwt
    access_token = create_access_token(data={"sub": {'email': user.email, 'id': user.id}})
    return {
        'message': 'Success',
        'status_code': 200,
        'data': {
            'email': user.email,
            'id': user.id
        },
        'access_token': access_token
    }


def show(user_id: int, response, db: Session = Depends(get_db)):
    user: models.User = db.query(models.User).filter(models.User.id == user_id).first()
    if not user:
        response.status_code = status.HTTP_404_NOT_FOUND
        return {
            'message': "Not found",
            'status_code': 404,
            'error': 'NOT FOUND'
        }
    return {
        'message': "success",
        'status_code': 200,
        'status': 'Success',
        'data': user
    }

Still in the app directory create another folder called helpers. create the following files inside init.py, jwt_helper.py for jwt encode and decode and password_helper.py for password hashing and verifying hashes.

password_helper.py file

from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


def hash_password(plain_password: str):
    return pwd_context.hash(plain_password)


def verify_password(plain_password: str, hashed_password: str):
    return pwd_context.verify(plain_password, hashed_password)

jwt_helper.py file

import os
from datetime import datetime, timedelta
from typing import Union
from passlib.context import CryptContext
from jose import JWTError
import jwt as pyjwt
from dotenv import load_dotenv

load_dotenv()
from app import schema

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
ACCESS_TOKEN_EXPIRE_MINUTES = 30


def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=60)
    to_encode.update({"exp": expire})
    encoded_jwt = pyjwt.encode(to_encode, os.environ.get('SECRET_KEY'), algorithm=os.environ.get('ALGORITHM'))
    return encoded_jwt


def verify_token(credentials, credentials_exception):
    try:
        token = credentials.credentials
        payload = pyjwt.decode(token, os.environ.get('SECRET_KEY'), algorithms=[os.environ.get('ALGORITHM')])
        user = payload.get('sub')
        if user is None:
            raise credentials_exception
        return schema.TokenData(id=user['id'], email=user['email'])
    except JWTError:
        raise credentials_exception

CRUD API

In the routers folder. create a file called todo_router.py and paste this code

from fastapi import Depends, APIRouter, Response
from sqlalchemy.orm import Session
from starlette import status

from app.database import get_db
from app import schema
from app.dependencies import get_current_user
from app.repository import todo_repository

router = APIRouter(prefix="/api/v1/todos", tags=["todos"], )


@router.post('/', status_code=201)
def create(request: schema.Todo, response: Response, db: Session = Depends(get_db),
           current_user: schema.User = Depends(get_current_user)):
    return todo_repository.create(request, response, db, current_user)


@router.get("/", status_code=status.HTTP_200_OK)
def fetch_todos(db: Session = Depends(get_db)):
    return todo_repository.fetch_todos(db)


@router.get("/{todo_id}", status_code=200)
def show(todo_id: int, response: Response, db: Session = Depends(get_db)):
    return todo_repository.show(todo_id, response, db)


@router.put("/{todo_id}", status_code=201)
def update_todo(todo_id: int, request: schema.Todo, response: Response, db: Session = Depends(get_db),
                current_user: schema.User = Depends(get_current_user)):
    return todo_repository.update(todo_id, request, response, db)


@router.delete("/{todo_id}", status_code=201)
def destroy(todo_id: int, response: Response, db: Session = Depends(get_db),  current_user: schema.User = Depends(get_current_user)):
    return todo_repository.destroy(todo_id, response, db)

In the repository folder, create a file called todo_repository.py and paste this code

from fastapi import Depends
from sqlalchemy.orm import Session, joinedload
from starlette import status
from app import models, schema
from app.database import get_db
from app.dependencies import get_current_user


def create(request: schema.Todo, response, db: Session = Depends(get_db),
           current_user: schema.User = Depends(get_current_user)):
    todo = db.query(models.Todo).filter(models.Todo.title == request.title).first()
    if todo:
        response.status_code = status.HTTP_409_CONFLICT
        return {
            'message': "todo already exist",
            'status_code': 409,
            'error': 'CONFLICT'
        }

    new_todo = models.Todo(title=request.title, user_id=current_user.id)
    db.add(new_todo)
    db.commit()
    db.refresh(new_todo)
    return {
        'message': "success",
        'status_code': 201,
        'status': 'Success',
        'data': {
            'id': new_todo.id,
            'email': new_todo.title
        }
    }


def fetch_todos(db: Session):
    todos = db.query(models.Todo).options(joinedload(models.Todo.user)).order_by(models.Todo.id.desc()) .all()
    return todos


def show(post_id: int, response, db: Session = Depends(get_db)):
    todo = db.query(models.Todo).options(joinedload(models.Todo.user)).filter(models.Todo.id == post_id) .first()
    if not todo:
        response.status_code = status.HTTP_404_NOT_FOUND
        return {
            'message': "Not found",
            'status_code': 404,
            'error': 'NOT FOUND'
        }
    return {
        'message': "success",
        'status_code': 200,
        'status': 'Success',
        'data': todo
    }


def update(todo_id, request: schema.Todo, response, db: Session = Depends(get_db)):
    todo = db.query(models.Todo).filter(models.Todo.id == todo_id).first()
    if not todo:
        response.status_code = status.HTTP_404_NOT_FOUND
        return {
            'message': "Not found",
            'status_code': 404,
            'error': 'NOT FOUND',
        }
    todo.title = request.title
    db.add(todo)
    db.commit()
    db.refresh(todo)
    return {
        'message': "successful",
        'status_code': 201,
        'status': 'Success',
        'data': todo
    }


def destroy(todo_id: int, response, db: Session = Depends(get_db)):
    todo = db.query(models.Todo).filter(models.Todo.id == todo_id).delete(synchronize_session=False)
    db.commit()
    if not todo:
        response.status_code = status.HTTP_404_NOT_FOUND
        return {
            'message': "Not found",
            'status_code': 404,
            'error': 'NOT FOUND',
        }
    return {
        'message': "Deleted successfully",
        'status_code': 200,
        'status': 'Success',
        'data': todo
    }

in the folder create a .env file

SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0"
ALGORITHM = "HS256"
SQLALCHEMY_DATABASE_URL = "postgresql://username:password@localhost/database_name"

Note:

• localhost:8000/docs to see API document

For more understanding, do well to this the official documentation

Link to the source code

Thank you for reading.