FastAPI is a modern web framework for Python that makes it easy to build APIs quickly. One of the key functionalities FastAPI provides is authentication, and OAuth2 is a popular approach for securing apps. In this guide, we'll explore how to implement OAuth2 authentication using Python’s FastAPI.

Step 1: Install Required Libraries

Before we get started, make sure you have FastAPI and an ASGI server (like uvicorn) installed. You might also want to install passlib for password hashing.

pip install fastapi[all] uvicorn passlib

Step 2: Set Up the Project Structure

Create a new directory for your FastAPI application and structure it like this:

/oauth2_fastapi
    └── main.py

Step 3: Create the FastAPI Application

Now, let’s create your FastAPI application in main.py. Start by importing the required modules and setting up your environment variables for OAuth2:

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from passlib.context import CryptContext

app = FastAPI()

Step 4: Create a Hashing Context

Using passlib, you can create a password hashing context:

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

Step 5: Define Your User Model

Next, define a User model that includes the username, password, and other fields as needed:

from pydantic import BaseModel

class User(BaseModel):
    username: str
    email: str
    full_name: str
    disabled: bool = None

Step 6: Set Up OAuth2 Configuration

Now you will define the OAuth2PasswordBearer, which will extract the token from the Authorization header.

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

Step 7: Create a Fake Database

For simplicity, let’s set up a fake user database. In a real application, this would be replaced by a proper database.

fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": pwd_context.hash("secret"),
        "disabled": False,
    }
}

Step 8: Create a Helper Function to Verify Passwords

To verify user credentials, we need a function that checks if the provided password matches the stored hashed password.

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

Step 9: Create User Authentication Logic

This function will authenticate a user based on the provided username and password:

def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return User(**user_dict)
    
def authenticate_user(db, username: str, password: str):
    user = get_user(db, username)
    if not user or not verify_password(password, user.hashed_password):
        return False
    return user

Step 10: Create a Token Endpoint

Define a route that generates a token when the user provides valid credentials:

from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt

SECRET_KEY = "your_secret_key_here"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

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})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

Step 11: Protect Your Routes

You should now protect your endpoints using the oauth2_scheme. Here’s an example of a protected route:

@app.get("/users/me")
async def read_users_me(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = get_user(fake_users_db, username)
    if user is None:
        raise credentials_exception
    return user

Step 12: Run Your Application

Finally, run your FastAPI application with uvicorn:

uvicorn main:app --reload

Now you can test your setup by making requests to the /token endpoint to get a bearer token, which you can then use to access the protected /users/me endpoint.

By integrating OAuth2 into your FastAPI application, you ensure secure access to your API endpoints. FastAPI’s intuitive design makes implementing OAuth2 straightforward with just a few lines of code. Happy coding!