‹ FastAPI Interview

FastAPI Basics FastAPI Uvicorn FastAPI Routing FastAPI Pydantic Models FastAPI Automatic API Documentation FastAPI Handling Request Data FastAPI SQLAlchemy ORM FastAPI Tortoise ORM FastAPI Dependency Injection FastAPI Response Models FastAPI Error Handling FastAPI OAuth2 FastAPI JWT (Json Web Tokens) FastAPI RBAC FastAPI Background Tasks FastAPI Middleware FastAPI Handling File Uploads FastAPI Async Programming FastAPI WebSockets FastAPI Testing FastAPI Security FastAPI Extensions FastAPI CORS FastAPI Deployment FastAPI Optimization FastAPI GraphQL FastAPI Task Queues And Celery FastAPI Custom CLI Commands FastAPI Server-Sent Events

FastAPI OAuth2

What is OAuth2 in FastAPI?

OAuth2 is an open standard for access delegation, commonly used as a way to grant websites or applications limited access to user accounts on other services. In FastAPI, OAuth2 is supported for handling authentication via access tokens. FastAPI simplifies the implementation of OAuth2 by providing built-in utilities for handling token generation, validation, and user authentication.

How does FastAPI implement OAuth2 with password flow?

FastAPI provides the OAuth2PasswordBearer class for implementing OAuth2 with password flow. The password flow allows a user to send their username and password to retrieve an access token, which can be used to authenticate API requests.

Example of implementing OAuth2 with password flow:

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.post("/token")
async def login():
    return {"access_token": "test_token", "token_type": "bearer"}

@app.get("/users/me")
async def read_users_me(token: str = Depends(oauth2_scheme)):
    if token != "test_token":
        raise HTTPException(status_code=401, detail="Invalid token")
    return {"user": "john"}

In this example, the route /token provides a token when a user logs in, and the /users/me route is protected by the OAuth2 token, requiring a valid token to access it.

What is the `OAuth2PasswordBearer` class in FastAPI?

The OAuth2PasswordBearer class is a utility provided by FastAPI to handle OAuth2 authentication with password flow. It returns the token sent by the client in the Authorization header and passes it to the route function for validation. It simplifies the process of securing routes by automatically handling token extraction and validation.

Example of using OAuth2PasswordBearer:

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.get("/secure-data")
async def secure_data(token: str = Depends(oauth2_scheme)):
    return {"token": token}

In this example, the token is extracted from the request's Authorization header and passed to the route function for further validation.

How do you generate access tokens in FastAPI using OAuth2?

To generate access tokens in FastAPI, you typically use a secure method like JWT (JSON Web Tokens). You generate a token when a user logs in and return it to the client, which uses it for subsequent authenticated requests.

Example of generating access tokens:

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

SECRET_KEY = "mysecret"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

@app.post("/token")
async def login():
    access_token = create_access_token(data={"sub": "user_id"})
    return {"access_token": access_token, "token_type": "bearer"}

In this example, the create_access_token function generates a JWT with an expiration time, which is then returned to the client in the /token route.

How do you verify access tokens in FastAPI using OAuth2?

To verify access tokens in FastAPI, you decode the token using a secret key and ensure that it is valid. You also check that the token has not expired and contains the required claims.

Example of verifying access tokens:

from jose import JWTError, jwt

def verify_token(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: str = payload.get("sub")
        if user_id is None:
            raise HTTPException(status_code=401, detail="Invalid token")
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user_id

@app.get("/users/me")
async def read_users_me(token: str = Depends(oauth2_scheme)):
    user_id = verify_token(token)
    return {"user_id": user_id}

In this example, the verify_token function decodes the JWT token and ensures it contains the sub claim (user ID). If the token is invalid, an HTTPException is raised.

How do you protect routes using OAuth2 in FastAPI?

To protect routes in FastAPI using OAuth2, you use the OAuth2PasswordBearer dependency in your route function. You can then validate the token and ensure that only authenticated users can access the route.

Example of protecting routes using OAuth2:

from fastapi import Depends, HTTPException

@app.get("/protected-route")
async def protected_route(token: str = Depends(oauth2_scheme)):
    if token != "test_token":
        raise HTTPException(status_code=401, detail="Invalid token")
    return {"message": "This is a protected route"}

In this example, the protected_route is secured using the OAuth2 token. If the token is invalid or missing, a 401 Unauthorized error is returned.

How do you refresh access tokens in FastAPI?

To refresh access tokens in FastAPI, you typically implement a separate route for refreshing tokens. This route generates a new access token using a refresh token that is provided by the client. The refresh token is usually stored securely and has a longer expiration time.

Example of refreshing access tokens:

@app.post("/refresh-token")
async def refresh_token(refresh_token: str):
    # Verify the refresh token
    if refresh_token == "valid_refresh_token":
        new_access_token = create_access_token(data={"sub": "user_id"})
        return {"access_token": new_access_token, "token_type": "bearer"}
    raise HTTPException(status_code=401, detail="Invalid refresh token")

In this example, the /refresh-token route verifies the refresh token and generates a new access token for the client.

How do you handle scopes with OAuth2 in FastAPI?

OAuth2 scopes are used to specify different levels of access for the client. You can define scopes in FastAPI to control what resources the client can access based on the token's scope. FastAPI automatically validates the scopes in the token.

Example of using OAuth2 scopes:

from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token", scopes={"read": "Read access", "write": "Write access"})

@app.get("/users/me", dependencies=[Depends(oauth2_scheme)])
async def read_users_me(token: str = Depends(oauth2_scheme)):
    # Check if the token has the necessary scope
    return {"user": "Authenticated with correct scope"}

In this example, the OAuth2 token is validated for specific scopes, such as "read" or "write" access.

How do you handle token expiration in FastAPI with OAuth2?

To handle token expiration in FastAPI, you define an expiration time when generating the token. When verifying the token, you ensure that the token has not expired. If the token is expired, the client is required to obtain a new one.

Example of handling token expiration:

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

SECRET_KEY = "mysecret"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    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])
        exp = payload.get("exp")
        if datetime.utcnow() > datetime.utcfromtimestamp(exp):
            raise HTTPException(status_code=401, detail="Token expired")
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

In this example, the token is created with an expiration time, and the verify_token function checks if the token has expired before allowing access.

How do you revoke tokens in FastAPI?

Token revocation is a process where an access token is invalidated before its expiration time. To revoke tokens in FastAPI, you would typically maintain a blacklist of revoked tokens and check against it during token verification. If the token is in the blacklist, it is considered revoked and access is denied.

Example of token revocation:

revoked_tokens = set()

@app.post("/revoke-token")
async def revoke_token(token: str):
    revoked_tokens.add(token)
    return {"message": "Token revoked"}

def verify_token(token: str):
    if token in revoked_tokens:
        raise HTTPException(status_code=401, detail="Token revoked")
    # Continue token verification process

In this example, the revoke-token route adds tokens to a revoked list, and the verify_token function checks whether the token is revoked before proceeding.

Ads