Securing FastAPI with OAuth2 and JWT

1. The Tool for the Job: OAuth2PasswordBearer

To start, we need to tell FastAPI how clients are expected to provide the token. The standard method is using an Authorization header with a "Bearer" scheme, which looks like this: Authorization: Bearer <your_long_token_string>.

FastAPI provides a utility class specifically for this: OAuth2PasswordBearer. Let's create an instance of it.

from fastapi.security import OAuth2PasswordBearer




# The tokenUrl is the URL that the client will use to get the token.
# In our case, it's the endpoint we built in the last lesson.
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

The tokenUrl parameter might seem a bit confusing at first. It does not mean that this object will send a request to that URL to validate the token. Instead, its primary purpose is for the OpenAPI documentation. When you use this scheme, Swagger UI will show an "Authorize" button. The tokenUrl tells the interactive docs which endpoint to use to acquire a token when you enter your username and password in the UI's authorization modal.

2. Creating the Core Validation Dependency

Now that we have our oauth2_scheme, we can use it to build our main dependency. This function will be responsible for extracting the token, decoding it, validating its contents, and fetching the corresponding user. Let's call it get_current_user.

The official FastAPI documentation provides the canonical implementation for this. It's a critical piece of code to understand.

Let's break down the function you just read about. Here is a typical implementation, combining the concepts from the documentation:

from typing import Annotated
from datetime import datetime

from fastapi import Depends, HTTPException, status
import jwt # PyJWT library
from pydantic import BaseModel

from . import crud, schemas # Your project's modules
from .utils import SECRET_KEY, ALGORITHM # From previous lesson




# This is the scheme we just defined
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

class TokenData(BaseModel):
    username: str | None = None

async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:



        # 1. Decode the token
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        



        # 2. Extract the username from the 'sub' claim
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
            
        token_data = TokenData(username=username)

    except jwt.PyJWTError:



        # This catches any error from PyJWT (e.g., invalid signature, expired token)
        raise credentials_exception
        



    # 3. Fetch the user from the database
    user = crud.get_user(username=token_data.username) # This function queries your DB
    if user is None:
        raise credentials_exception
        



    # 4. Return the user object (e.g., a Pydantic model)
    return user

Let's dissect this powerful function:

1. token: Annotated[str, Depends(oauth2_scheme)]: This is the dependency injection in action.
   • Depends(oauth2_scheme) tells FastAPI to execute our oauth2_scheme instance.
   • This instance looks for the Authorization header, checks if it's a Bearer token, and if so, passes the token string to our token parameter.
   • If no token is found, or the header is malformed, it automatically raises a 401 error. Your function's code won't even run.
2. jwt.decode(...): We use the PyJWT library to decode the token. This also verifies the signature and the expiration time. If either is invalid, it raises an error.
3. try...except jwt.PyJWTError: This is crucial for security. We wrap the decoding in a try...except block. Any failure in decoding means the token is invalid, so we immediately raise our credentials_exception.
4. Payload and User Validation: After decoding, we extract the user identifier (from the sub claim) and use it to fetch the user from our database. If the user no longer exists, we again raise the credentials_exception.
5. Return User: If all checks pass, the function returns the user object. FastAPI will then inject this return value into the path operation function wherever it's requested.

For a dynamic walkthrough of this process, the following video is an excellent supplement.

3. Protecting an Endpoint

Now that we've built our guard, let's post it at a door. To protect a path operation, you simply add get_current_user as another dependency.

Imagine we want to create an endpoint /users/me that returns the profile of the currently logged-in user.

@router.get("/users/me", response_model=schemas.User)
async def read_users_me(
    current_user: Annotated[schemas.User, Depends(get_current_user)]
):
    """
    Get the profile for the current user.
    """
    return current_user

This is where the elegance of FastAPI's dependency injection shines:

• Before running read_users_me, FastAPI first runs get_current_user.
• If get_current_user raises an HTTPException, the request is immediately terminated with a 401 Unauthorized response. The code inside read_users_me is never executed.
• If get_current_user succeeds, it returns a user object. FastAPI assigns this object to the current_user parameter.
• Your path operation function can now simply use the current_user object, confident that the user is authenticated.

This simple act of adding the dependency is what puts the padlock icon next to your endpoints in the Swagger UI.

This pattern is incredibly powerful. The dependency handles the complex and repetitive logic of authentication, keeping your actual business logic clean and focused.

Conclusion

Today you've implemented the most critical part of your API's security. You now have a reusable dependency that can be used to protect any endpoint.

Key Takeaways:

• OAuth2PasswordBearer is a utility class that creates a dependency for extracting a Bearer token from the Authorization header.
• The tokenUrl parameter is used by the interactive API docs to know where to request a token.
• A validation dependency (e.g., get_current_user) is an async def function that uses Depends(oauth2_scheme) to get the token.
• Inside this dependency, you must decode the token, handle potential errors with a try...except block, and verify the user exists in your database.
• Invalid tokens or users should trigger a 401 Unauthorized HTTPException.
• To protect an endpoint, you add this validation function as a dependency to the path operation's signature. The return value of the dependency is injected as an argument.

Preview of the Next Lesson:

We've now built the get_current_user dependency and seen how to use it. In the next lesson, we will focus on protecting routes by injecting the current authenticated user via a dependency and formally compare this pattern to DRF's permission_classes. We'll explore how to chain dependencies to create more granular checks, such as verifying if a user is active or has specific permissions, which will set us up for implementing role-based access control.