Tutorial #52: Implementing OAuth2 Authentication in FastAPI with SQLModel

In this tutorial, we will be integrating OAuth2 authentication in FastAPI using SQLModel. OAuth2 is a secure and industry-standard protocol for authorization that is widely used in modern web applications. SQLModel is a library that allows you to interact with databases using Python data models, which makes it a perfect fit for integrating with FastAPI.

To get started, make sure you have FastAPI and SQLModel installed in your Python environment. You can install them using pip:

pip install fastapi
pip install sqlalchemy
pip install python-multipart
pip install sqlmodel

Next, create a new Python file for our FastAPI app. Let’s call it main.py. In this file, we will define our FastAPI application and set up the OAuth2 authentication using SQLModel.

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from .schemas import TokenData
from .oauth2 import get_current_user

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.get("/items/")
async def read_items(token: str = Depends(oauth2_scheme)):
    return {"token": token}

In the code above, we have created a basic FastAPI application with a single endpoint /items/. This endpoint requires an OAuth2 token for authentication, which will be sent in the Authorization header.

Next, let’s define the Token model and some helper functions for working with OAuth2 tokens.

from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext

SECRET_KEY = "someverysecuresecretkey"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

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

class TokenData:
    username: str = None

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=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

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

In the code above, we have defined the TokenData class to hold the username parsed from the token. We have also defined helper functions for creating and verifying OAuth2 tokens using the jwt library.

Next, let’s define the User model and a fake database for testing our authentication system.

from sqlmodel import SQLModel, create_engine, Session, select
from sqlmodel.ext.asyncio.session import AsyncSession
import asyncio

DATABASE_URL = "sqlite:///./test.db"

class User(SQLModel, table=True):
    id: int = None
    username: str
    hashed_password: str

async def get_user(username: str):
    async with AsyncSession() as session:
        statement = select(User).where(User.username == username)
        result = await session.execute(statement)
        return result.scalars().first()

def get_password_hash(password: str):
    return pwd_context.hash(password)

async def create_user(username: str, password: str):
    async with AsyncSession() as session:
        user = User(username=username, hashed_password=get_password_hash(password))
        session.add(user)
        await session.commit()
        await session.refresh(user)
        return user

In the code above, we have defined the User model with fields for id, username, and hashed_password. We have also defined helper functions for getting a user by username, hashing passwords, and creating a new user in the fake database.

Next, let’s define the authentication dependencies and functions for our FastAPI endpoints.

from fastapi.security import OAuth2PasswordBearer

def oauth2_scheme():
    return OAuth2PasswordBearer(tokenUrl="token")

def authenticate_user(username: str, password: str):
    user = get_user(username)
    if not user:
        return False
    if not pwd_context.verify(password, user.hashed_password):
        return False
    return user

@app.post("/token")
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if user is False:
        raise HTTPException(status_code=400, detail="Incorrect username or password")
    access_token = create_access_token(data={"sub": user.username})
    return {"access_token": access_token, "token_type": "bearer"}

In the code above, we have defined the oauth2_scheme dependency for requiring OAuth2 tokens in our endpoints. We have also defined a function for authenticating users using their username and password, as well as a /token endpoint for generating access tokens.

Finally, let’s run our FastAPI application.

if __name__ == "__main__":
    import uvicorn
    from .database import create_database
    create_database()
    uvicorn.run(app, host="0.0.0.0", port=8000)

And that’s it! We have successfully integrated OAuth2 authentication in FastAPI using SQLModel. You can now test your endpoints using tools like Postman or curl. Remember to always protect your tokens and keep your authentication system secure.