An Introduction to FastAPI

Pritam Ghosh — Wed, 25 Jan 2023 10:11:59 +0000

FastAPI is a modern, very fast (high-performance), web framework for building RESTful APIs. Here we discuss its features and usage briefly by creating a simple Blog app for better understanding.

Features:

Some of the big tech companies like Netflix, Microsoft, Uber and so on are also using FastAPI.
The main features are,

FastAPI supports asynchronous endpoints by default, which allows Python developers to write concurrent code. This is a huge advantage over Flask, Django.
Makes use of Python 3.6 type declarations that allows you to specify the type of a variable, Autocompletion works amazingly.
Security and authentication integrated like - HTTP Basic, OAuth2 (including JWT tokens), pass API keys in Headers or in Query params.
Supports an extremely powerful Dependency Injection system, dependencies can be nested (dependencies can have dependencies) as well.
As it is based on Starlette, we can use all its features like - Startup and shutdown events, Session, and Cookie, WebSocket, CORS, GZip, Static Files, Streaming responses etc.
It can validate request data, you can have deeply nested JSON objects and have them all validated and annotated by Pydantic.
Automatic data model documentation with JSON Schema (as OpenAPI itself is based on JSON Schema). Supports automatic swagger UI documentation for APIs.

FastAPI depends on three major tools:

Starlette: A lightweight ASGI (Asynchronous Server Gateway Interface) HTTP web framework.

Pydantic: A Python library for data validation, request parsing and response model. It has efficient error handling and a custom validation mechanism.
Below is a pydantic Model for response sending. We will use it later with some practical examples.

from pydantic import BaseModel

class PostResponse(BaseModel):
    id: int
    title: str
    body: str | None = None

    class Config:
        orm_mode = True

Uvicorn: An ASGI web server implementation. It includes a Gunicorn worker class allowing you to run ASGI applications, with all Uvicorn's performance benefits.

def main():
    uvicorn.run(
        app,
        host="0.0.0.0",
        port=8000,
        log_level="info"
    )

if __name__ == "__main__":
    main()

Let’s create a simple Blog app:

Install:
a. Python 3.6+,
b. PostgreSQL
create virtualenv: python -m venv env
activate virtualenv: source env/bin/activate
c. Dependencies: pip install fastapi uvicorn[standard] psycopg2 sqlalchemy
Add a main.py file in project folder which will be the entry point of the app and include these lines, here we,
a. Instantiate the app object of FastAPI
b. Made the connection with postgresql database
c. And, included the router of API endpoints

import uvicorn
from fastapi import FastAPI

from .views import router
from .models import Base

app = FastAPI()
app.include_router(router, prefix="/api")
Base.metadata.create_all(bind=engine)

def main():
    uvicorn.run(
app, host="0.0.0.0", port=8000, log_level="info"
)

if __name__ == "__main__":
    main()

Add all the database configuration related things in database.py file

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "postgresql+psycopg2://username:password@localhost:5432/blog"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

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

Create a sqlalchemy model name Post with fields(id, title, body) where we store our posts of Blog and a pydantic model (use the above mentioned pydantic model) inherited from pydantic BaseModel in models.py file.

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text

Base = declarative_base()

class Post(Base):
    __tablename__ = "post"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String)
    body = Column(Text)

Let’s add an API endpoint to save a Post using the post method in views.py file, this is a simple synchronous call. To make it an asynchronous call we just need to add an async statement before the def. FastAPI is based on AnyIO which makes it compatible with both Python's standard library asyncio. Here, we use dependency injection to make the endpoint depend on our database connection which itself is a sqlalchemy session type.

Before look into the API just a short comparison on sync
and async Python:

def sync_function():
    results = blocking_function()
    return results

async def async_function():
    results = await non_blocking_awaitable()
    return results

Now, it will be easy to understand the REST endpoints

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session

from .database import get_db
from .models import Post, PostResponse

router = APIRouter()

@router.post("/posts")
def add_post(post_data, db: Session = Depends(get_db)):
    post = Post(title=post_data.title, body=post_data.body)
    db.add(post)
    db.commit()
    return post

@router.post("/posts")
async def add_post(post_data, db: Session = Depends(get_db)):
    ...

Use Postman or FastAPI swagger to test APIs.

Now, retrieve the Post we just saved in the database using the get method. Here we’re using the pydantic model to structure the response json based.

@router.get("/posts/{post_id}", response_model=PostResponse)
def read_post(post_id, db: Session = Depends(get_db)):
    post_obj = db.query(Post).filter(Post.id == post_id).one_or_none()
    if not post_obj:
        raise HTTPException(status_code=400, detail="Post does not exist with this Id")
    return post_obj

Similarly, below are APIs to update and delete Post.

@router.put("/posts/{post_id}", response_model=Post)
def update_post(post_id, post_data, db: Session = Depends(get_db)):
    post_obj = db.query(Post).filter(Post.id == post_id).one_or_none()
    if not post_obj:
        raise HTTPException(status_code=400, detail="Post does not exist with this Id")
    post_obj.title = post_data.title
    post_obj.body = post_data.body
    db.commit()
    return post_obj

We can use the patch method for partial updates.

@router.delete("/posts/{post_id}")
def delete_post(post_id, db: Session = Depends(get_db)):
    post_obj = db.query(Post).filter(Post.id == post_id).one_or_none()
    if not post_obj:
        raise HTTPException(status_code=400, detail="Post does not exist with this Id")
    db.delete(post_obj)
    db.commit()
    return {"Post Deleted": True}

Advanced Topics:

Events Handling:

@app.on_event("startup")
async def startup():
    # connect any resources like database, redis etc...
    resource.connect()

@app.on_event("shutdown")
async def shutdown():
    # disconnect resources
    resource.disconnect()

Middleware usage: When we need to do some stuff before and after the response like authentication, authorization, process time and many more.

from starlette.middleware.base import BaseHTTPMiddleware
from fastapi.responses import Response

class CustomMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next) -> Response:
        response = await call_next(request)
        ...

# Include this line in main.py file
app.add_middleware(CustomMiddleware)

FastAPI comes with powerful security features in the fastapi.security module,

from fastapi.security import (
    OAuth2PasswordBearer,
    OAuth2PasswordRequestForm,
    SecurityScopes,
)

By now, I have covered the best possible way to use the Framework and its features including async, typing, dependency injection, response model, events handling etc.

Some reference links:

https://fastapi.tiangolo.com/
https://codingnomads.co/blog/python-fastapi-tutorial

Conclusion:

The framework is now fully production-ready, with excellent documentation, support and easy to use. I hope you liked it and Thanks for your time.
HAPPY CODING...

Disclaimer:

This is a personal blog. The views and opinions expressed here are only those of the author and do not represent those of any organization or any individual with whom the author may be associated, professionally or personally.

DEV Community: Pritam Ghosh