Published on

Build and Secure an API in Python with FastAPI

Authors

Introduction :

As Python grows in popularity, the variety of high-quality frameworks available to developers has blossomed. In addition to steadfast options like Django and Flask, there are many new options including FastAPI.

First released in late 2018, FastAPI differentiates itself from other Python frameworks by offering a modern, fast, and succinct developer experience for building reliable REST APIs. While one of the newer open-source Python frameworks available, FastAPI has quickly gained a following with over 33,000 stars on GitHub and an active community of maintainers working on the project.

Our article is based on Creating an API with high performance built with FastAPI & SQLAlchemy, helps to improve connection with your Backend Side and stay relate using SQLite3 & a secure Schema Based on Python-Jose a JavaScript Object Signing and Encryption implementation in Python.

Getting Started

Initial Setup

Start by creating a new folder to hold your project called "SecureAPI":

$ mkdir SecureAPI
$ cd SecureAPI
  • Next, create and activate a virtual environment:

Feel free to swap out virtualenv and Pip for Poetry or Pipenv.

  • Create a file requirements.txt and add the Preconfigured packages relate to the project
fastapi
uvicorn
sqlalchemy
passlib
bcrypt
python-jose
python-multipart
  • I use pipenv a global python project pip install pipenv.
  • Create a virtual environment for this project
# creating pipenv environment for python 3
$ pipenv --three

# activating the pipenv environment
$ pipenv shell

# if you have multiple python 3 versions installed then
$ pipenv install -d --python 3.8

# install all dependencies (include -d for installing dev dependencies)
$ pipenv install -d
  • Next Create the Following Files and Folders :
/---SecureAPI
|   .env
|   main.py
|   requirements.txt
+---api
|   |   blog.py
|   |   user.py
+---core
|   |   auth.py
|   |   blog.py
|   |   user.py
+---database
|   |   configuration.py
+---models
|   |   models.py
+---schema
|   |   hash.py
|   |   oa2.py
|   |   schemas.py
|   |   token.py

What we are building?

  • We are Building an API with high performance built with FastAPI & SQLAlchemy, help to improve connection with your Backend Side to create a simple blog and Cruds with OAuth2PasswordBearer ⛏.

  • Let's Code :

Database :

  • To Provide a good and fast work, i choose SQLAlchemy as a toolkit and performance, for the Database i choose SQLite a High performance Database and full of Features. Lets Create the Configuration.py:
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = 'sqlite:///blog.db'
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={
                       "check_same_thread": False})
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)
Base = declarative_base()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
  • If you want to configure the Database with an other Provider like MySQL or PostgreSQL you can change the Database_URL here :
# here you need to inster the  URI that should be used for the connection.
SQLALCHEMY_DATABASE_URL = 'sqlite:///blog.db'
  • For Example to :
SQLALCHEMY_DATABASE_URL = 'mysql://username:password@server/blog'

Models :

  • Here in the models.py, we are gonna create 2 tables based on the requirements related to this project blogs and users :

  • Lets declare our imports first :

from database.configuration import Base
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship
  • The Blogs Table :
class Blog(Base):
    __tablename__ = "blogs"
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String)
    body = Column(String)
    user_id = Column(Integer, ForeignKey("users.id"))
    creator = relationship("User", back_populates="blogs")
  • The Users Table :
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String)
    email = Column(String)
    password = Column(String)
    blogs = relationship("Blog", back_populates="creator")

schemas :

  • Here we are declaring Blogs and Users and Token models, it will contain all format int or string or bool, lets code this on our schemas.py :

  • declare all pydantic & Typing imports :

from pydantic import BaseModel
from typing import Optional, List
from pydantic.main import BaseConfig
  • Our Schema gonna be like this :
class BlogBase(BaseModel):
    title: str
    body: str

class Blog(BlogBase):
    class Config():
        orm_mode = True

class User(BaseModel):
    name: str
    email: str
    password: str

class ShowUser(BaseModel):
    name: str
    email: str
    blogs: List[Blog] = []

    class Config():
        orm_mode = True

class ShowBlog(BaseModel):
    title: str
    body: str
    creator: ShowUser

    class Config():
        orm_mode = True

class Login(BaseModel):
    username: str
    password: str

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

class TokenData(BaseModel):
    username: Optional[str] = None

API :

  • Now we need to code the Main API and create the Queries with a hashed Password for the Users using Bcrypt:

  • User.py:

from sqlalchemy.orm import Session
from schema import schemas
from models import models
from fastapi import HTTPException, status
from schema.hash import Hash

def create(request: schemas.User, db: Session):
    hashedPassword = Hash.bcrypt(request.password)
    user = models.User(name=request.name, email=request.email,
                       password=hashedPassword)
    db.add(user)
    db.commit()
    db.refresh(user)
    return user

def show(id: int, db: Session):
    user = db.query(models.User).filter(models.User.id == id).first()
    if not user:
        raise HTTPException(status.HTTP_404_NOT_FOUND,
                            detail=f"User with id {id} not found")
    return user

def get_all(db: Session):
    return db.query(models.User).all()
  • Now lets Create the APi's for the blog where we gonna add the Requirements like Create, update, Show, Delete.
  • Blog.py:
from sqlalchemy.orm import Session
from schema import schemas
from models import models
from fastapi import HTTPException, status

def get_all(db: Session):
    blogs = db.query(models.Blog).all()
    return blogs

def create(request: schemas.Blog, db: Session):
    new_blog = models.Blog(title=request.title, body=request.body, user_id=1)
    db.add(new_blog)
    db.commit()
    db.refresh(new_blog)
    return new_blog

def destroy(id: int, db: Session):
    blog_to_delete = db.query(models.Blog).filter(models.Blog.id == id)

    if not blog_to_delete.first():
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
                            detail=f"Blog with id {id} not found.")
    blog_to_delete.delete(synchronize_session=False)
    db.commit()
    return {'done'}

def update(id: int, request: schemas.Blog, db: Session):
    blog = db.query(models.Blog).filter(models.Blog.id == id)
    if not blog.first():
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
                            detail=f"Blog with id {id} not found")
    blog.update(request.__dict__)
    db.commit()
    return 'updated'

def show(id: int, db: Session):
    blog = db.query(models.Blog).filter(models.Blog.id == id).first()
    if blog:
        return blog
    else:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
                            detail=f"Blog with the id {id} is not available")

Core :

  • In this folder we gonna Create 3 files Auth.py and Blog.py and User.py, all of this files are the routes for our API.
  • We are gonna Start by User.py, where we Create a routes for create_user, get_users, get_user_by_id.
  • User.py:
from fastapi import APIRouter, Depends, Response, status
from sqlalchemy.orm import Session
from schema import schemas
from models import models
from database import configuration
from typing import List
from api import user

router = APIRouter(tags=["Users"], prefix="/users")
get_db = configuration.get_db

# Create User
@router.post("/", status_code=status.HTTP_201_CREATED, response_model=schemas.ShowUser)
def create_user(request: schemas.User, db: Session = Depends(get_db)):
    return user.create(request, db)

# Get Users
@router.get("/", status_code=status.HTTP_200_OK, response_model=List[schemas.ShowUser])
def get_users(db: Session = Depends(get_db)):
    return user.get_all(db)

# Get Users using the ID
@router.get("/{id}", status_code=status.HTTP_200_OK, response_model=schemas.ShowUser)
def get_user_by_id(id: int, db: Session = Depends(get_db)):
    return user.show(id, db)
  • Now we need to add the Blog.py, relate to the API where we provide all Requirements.
  • Blog.py:
from schema.oa2 import get_current_user
from fastapi import APIRouter, Depends, status, Response
from schema import schemas
from database import configuration
from typing import List
from sqlalchemy.orm import Session
from api import blog

router = APIRouter(tags=["Blogs"], prefix="/blog")
get_db = configuration.get_db

@router.get("/", response_model=List[schemas.ShowBlog])
def get_all_blogs(db: Session = Depends(get_db), current_user: schemas.User = Depends(get_current_user)):
    return blog.get_all(db)

@router.post("/", status_code=status.HTTP_201_CREATED)
def create(request: schemas.Blog, db: Session = Depends(get_db), current_user: schemas.User = Depends(get_current_user)):
    return blog.create(request, db)

@router.get("/{id}", status_code=status.HTTP_200_OK, response_model=schemas.ShowBlog)
def get_blog_by_id(id: int, response: Response, db: Session = Depends(get_db), current_user: schemas.User = Depends(get_current_user)):
    return blog.show(id, db)

@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_blog(id: int, db: Session = Depends(get_db), current_user: schemas.User = Depends(get_current_user)):
    return blog.destroy(id, db)

@router.put("/{id}", status_code=status.HTTP_202_ACCEPTED)
def update_blog(id: int, request: schemas.Blog, db: Session = Depends(get_db), current_user: schemas.User = Depends(get_current_user)):
    return blog.update(id, request, db)
  • Here we mention some Route for Authentication login, that why we create the Auth.py to create a route relate to login for the user.
  • Auth.py :
from fastapi.security.oauth2 import OAuth2PasswordRequestForm
from schema.token import ACCESS_TOKEN_EXPIRE_MINUTES, create_access_token
from datetime import timedelta
from fastapi import APIRouter, Depends, HTTPException, status
from schema import schemas
from models import models
from database import configuration
from sqlalchemy.orm import Session
from schema.hash import Hash

router = APIRouter(prefix="/login",tags=["Authentication"])

@router.post("/")
def login(request: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(configuration.get_db)):
    user: schemas.User = db.query(models.User).filter(
        models.User.email == request.username).first()
    if not user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
                            detail=f"Invalid Credentials")
    if not Hash.verify(user.password, request.password):
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
                            detail=f"Incorrect password")
    # access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.email}
    )
    # generate JWT token and return
    return {"access_token": access_token, "token_type": "bearer"}

about from schema.hash import Hash we gonna describe it on the Security Side.

  • Now we Create all our routes we need to Secure them that why i use fastapi.security to implement and use OAuth2PasswordBearer for Authentication.

Security :

  • Back to the schema folder we need to describe and code some files relate to JWT "Json Web Token" using The bcrypt also to hash all password relate to login and Authentication.
  • hash.py:
from passlib.context import CryptContext

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

class Hash():
    @staticmethod
    def bcrypt(password: str):
        return pwd_ctx.hash(password)
    def verify(hashed_password, plain_password):
        return pwd_ctx.verify(plain_password, hashed_password)
  • Now, we need to code our token.py after define it on the schemas file as a Tokendata, using Python-jose to create_access_token and verify_token also for generate a Secret_Key and put it into the .env file.
  • Token.py :
from typing import Optional
from datetime import timedelta, datetime
from jose import jwt, JWTError
from schema.schemas import TokenData
import os

basedir = os.path.abspath(os.path.dirname(__file__))

SECRET_KEY = str(os.environ.get("SECRET_KEY"))
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = str(os.environ.get("ACCESS_TOKEN_EXPIRE_MINUTES"))

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

def verify_token(token: str, credentials_exception):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=ALGORITHM)
        email: str = payload.get("sub")
        if email is None:
            raise credentials_exception
        token_data = TokenData(email=email)
    except JWTError:
        raise credentials_exception

To generate a SECRET_KEY u can use openssl rand -hex 32.

  • Now we need to write our SECRET_KEY and ACCESS_TOKEN_EXPIRE_MINUTES into the enviromment variables file :
  • .env:
SECRET_KEY = <Generate your Secret_key>
ACCESS_TOKEN_EXPIRE_MINUTES = <estimate Time "int">
  • We didn't finish yet, we need now to write a oa2.py is the OAuth 2.0 an industry-standard protocol for authorization.
  • This file is relate to our Token cause when we get_current_user we need to verify the token after login when we use oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login").
  • oa2.py :
from fastapi import HTTPException, status, Depends
from fastapi.security import OAuth2PasswordBearer
from schema.token import verify_token

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "bearer"},)
    return verify_token(token, credentials_exception=credentials_exception)
  • Now after creating all Requirements files and folders we need to run the project, that why we need to code the main.py where we gonna Create the database tables and Create the instance also includes all Routes.
  • Here we gonna import all the basics files to run the Project :
from fastapi import FastAPI
from starlette.responses import HTMLResponse
# import locale files
from models import models
from database.configuration import engine
# import router files
from core import blog, user, auth
  • Now lets Create the database tables & the instance also includes all Routes :
models.Base.metadata.create_all(bind=engine)

app = FastAPI(
    title="SecureAPI",
    description="Hello thank you for reading my Blog UwU",
    version="1.0.0",)

app.include_router(blog.router)
app.include_router(user.router)
app.include_router(auth.router)
  • By default FastAPI return the response using JSONResponse, but we will Custom our Response using starlette.responses and HTMLResponse:
@app.get("/", response_class=HTMLResponse)
def index():
    return """
<!Doctype html>
    <html>
        <body>
            <h1>SecureAPI</h1>
            <div class="btn-group">
                <a href="/docs"><button>SwaggerUI</button></a>
                <a href="/redoc"><button>Redoc</button></a>
            </div>
        </body>
    </html>
"""
  • With this we can return a simple index with 2 buttons can redirect us to visualize and interact with the API’s resources without having any of the implementation logic in place.
  • Read more Here about SWAGGER UI or Redoc.

Running the Application

  • To run the Main.py we need to use uvicorn a lightning-fast ASGI server implementation, using uvloop and httptools.
# Running the application using uvicorn
$ uvicorn main:app --reload
  • Voila! Now the API should be Secure and work in a high level ❤
  • Try Now to play and interact with the enviromment to get more knowledges about APIs and also FastAPI.

Conclusion

  • This Article covered the process of Buidling and Securing a FastAPI application with JSON Web Tokens. You can find the Simular Source Code under the Name of Doge API Yezz123/DogeAPI. Thanks for reading.
Discuss on Twitter • View on GitHub