[Python] FastAPI authentication injection

My application has an AuthenticateService implemented as follows:

from domain.ports.repositories import ISalesmanRepository
from fastapi import HTTPException, status
from fastapi_jwt_auth import AuthJWT
from fastapi_jwt_auth.exceptions import JWTDecodeError
from shared.exceptions import EntityNotFound
from adapters.api.authentication.config import User


class AuthenticateService:
def __init__(self, user_repo: ISalesmanRepository):
self._repo = user_repo

def __call__(self, auth: AuthJWT) -> User:
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
auth.jwt_required()
user_id = auth.get_jwt_subject()
except JWTDecodeError:
raise credentials_exception

try:
user = self._repo.get_by_id(user_id)
return user
except EntityNotFound:
raise credentials_exception


So the behavior is basically:

• Is jwt is valid, get user from repository and returns
• Raise 401 if jwt is invalid

The problem is that in every controller implemented I have to repeat the process. I tried to implement a decorator that injects the user into the controller in case of success but I couldn't. I'm sure that the best way to implement this is to use fastAPI's Depends dependency injector. Today, a controller looks something like this:

from typing import Optional

from adapters.api.services import authenticate_service, create_sale_service
from fastapi import APIRouter, Depends
from fastapi_jwt_auth import AuthJWT
from pydantic import BaseModel

router = APIRouter()


class Request(BaseModel):
code: str
value: float
date: str
status: Optional[str] = None


@router.post('/sale')
def create_sale(request: Request, auth: AuthJWT = Depends()):
user = authenticate_service(auth)
result = create_sale_service.handle(
{"salesman": user, "sale": request.dict()}
)
return result.dict()


How can I abstract my authentication so that my controllers look like any of the versions below:


# Option 1: decorator

@router.post('/sale')
@authentication_required
def create_sale(request: Request, user: User): # User is the `__call__` response from `AuthenticateService` class
result = create_sale_service.handle(
{"salesman": user, "sale": request.dict()}
)
return result.dict()


# Option 2:
@router.post('/sale')
def create_sale(request: Request, user: User = Depends(authenticate_service)): # Something like that, using the depends to inject User to me
result = create_sale_service.handle(
{"salesman": user, "sale": request.dict()}
)
return result.dict()

Continue reading...