Building Scalable APIs with FastAPI 🚀

PythonAWSDockerFastAPI

Thursday, January 11, 2024

Introduction: Why FastAPI is a Great Choice for Modern APIs

So, why choose FastAPI over other frameworks? 🤔 Here are a few compelling reasons:

  • Speed: FastAPI is built on standard Python type hints and leverages the ASGI (Asynchronous Server Gateway Interface) framework, making it one of the fastest Python frameworks available. ⚡️
  • Robustness: FastAPI provides automatic interactive documentation, robust support for asynchronous programming, and built-in support for WebSockets.
  • Ease of use: FastAPI has a low barrier to entry, making it an ideal choice for developers of all skill levels.

🛠️ Setting Up FastAPI: Installing and Creating a Basic API

Here is a basic diagram of how our applications it's going to look like

fastapi diagram

Getting started with FastAPI is straightforward. Here's a step-by-step guide:

Installing FastAPI

You can install FastAPI using pip:

pip install fastapi

Creating a Basic API

Create a new file called main.py and add the following code:

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

Run the application using:

uvicorn main:app --reload

Open your web browser and navigate to http://127.0.0.1:8000/ to see the API in action. 🎉

Optimizing Performance: Async Operations and Database Connections

One of the coolest things about FastAPI is its built-in async support. It’s perfect for handling things like database queries, external API calls, and any other tasks that don’t need to block the main thread.

Using Async Functions for Non-Blocking Execution:

You can use async and await to make sure your app doesn’t get stuck waiting for slow tasks. Here's a simple example:

@app.get("/items/{item_id}")

async def read_item(item_id: int):

    return {"item": item_id}

Connecting to Databases:

FastAPI works great with databases! Whether you’re using SQLAlchemy (PostgreSQL, MySQL) or MongoDB, it’s super easy to set up.

SQLAlchemy (PostgreSQL, MySQL):

from sqlalchemy import create_engine

from sqlalchemy.orm import sessionmaker

DATABASE_URL = "postgresql://user:password@localhost/dbname"

engine = create_engine(DATABASE_URL)

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

MongoDB with Motor (MongoDB’s async driver):

from motor.motor_asyncio import AsyncIOMotorClient

client = AsyncIOMotorClient("mongodb://localhost:27017")

db = client.mydatabase

Authentication & Security

Security is super important, and FastAPI makes it easy to add authentication and protect your APIs. 🛡️

JWT Authentication:

JSON Web Tokens (JWT) are a popular way to securely transmit information between the client and server.

from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.get("/users/me")

async def read_users_me(token: str = Depends(oauth2_scheme)):
    return {"token": token}

OAuth2 Authentication (Github Example):

FastAPI lets you integrate OAuth2 with third-party providers like Google, GitHub, and more. Easy peasy!

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
import requests
import json

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

class User(BaseModel):
    username: str
    email: str

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

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # Change it for your credentials
    client_id = "your_client_id"
    client_secret = "your_client_secret"
    code = form_data.password
    
    # Exchange authorization code for access token
    token_url = "https://github.com/login/oauth/access_token"
    headers = {"Accept": "application/json"}
    data = {"client_id": client_id, "client_secret": client_secret, "code": code}
    response = requests.post(token_url, headers=headers, data=data)
    
    if response.status_code != 200:
        raise HTTPException(status_code=401, detail="Invalid credentials")
    
    # get the access token from the response
    access_token = response.json()["access_token"]
    
    # use the access token to get the user's information
    user_url = "https://api.github.com/user"
    headers = {"Authorization": f"Bearer {access_token}"}
    response = requests.get(user_url, headers=headers)
    
    # check if the response was successful
    if response.status_code != 200:
        raise HTTPException(status_code=401, detail="Invalid credentials")
    
    # get the user's information from the response
    user_data = response.json()
    username = user_data["login"]
    email = user_data["email"]
    
    # return the access token and user information
    return {"access_token": access_token, "token_type": "bearer", "username": username, "email": email}

Rate Limiting

You can use rate limiting to prevent abuse:

# Simple rate limiter
rate_limiter: Dict[str, int] = {}

def is_rate_limited(ip_address: str) -> bool:
    if ip_address in rate_limiter:
        if rate_limiter[ip_address] >= 5:
            return True
        else:
            rate_limiter[ip_address] += 1
    else:
        rate_limiter[ip_address] = 1
    return False

# use case
if is_rate_limited(ip_address):
    raise HTTPException(status_code=429, detail="Too many requests")

Deployment

You can deploy FastAPI using docker:

  1. Create docker file
FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .

RUN pip install -r requirements.txt

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
  1. Run & build a docker container
docker build -t fastapi-app .
docker run -p 8000:8000 fastapi-app
  1. Deploy to AWS Lambda

Want to go serverless? FastAPI works perfectly with AWS Lambda. You can use Mangum, a lightweight adapter to run FastAPI with Lambda.

First, let's install Mangum

pip install mangum

And here's how you can use it:


from fastapi import FastAPI, Request
from mangum import Mangum

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

handler = Mangum(app)

def lambda_handler(event, context):
    return handler(event, context)

And that's it, remember to add create your requirements.txt file!