24 : Authorization/Permissions in Fastapi


 Authorization and authentication are 2 different topics. Authentication is related to login and authorization is related to permission. Even if a person is logged in he/she may not have the necessary permissions. Consider our job-board has 3 admins. Now, anyone who knows our endpoints may make a put request and change our post! Even worse if that person puts a delete request for each of our job post! 😭
So, we need to verify if the person making the request has the necessary permissions. In our case, we want only the job posters or the superuser to be able to modify or delete the post. Enough talk lets see code. Before that, do you remember dependencies? We had created a dependency get_db which allows us to supply database connection to each request. This time also, we are going to create a dependency to identify current_user. fapis > version1 > route_login.py

from datetime import timedelta

from core.config import settings
from core.hashing import Hasher
from core.security import create_access_token
from db.repository.login import get_user
from db.session import get_db
from fastapi import APIRouter
from fastapi import Depends
from fastapi import HTTPException
from fastapi import status
from fastapi.security import OAuth2PasswordRequestForm,OAuth2PasswordBearer
from jose import JWTError, jwt
from schemas.tokens import Token
from sqlalchemy.orm import Session

router = APIRouter()

@router.post("/token", response_model=Token)
def login_for_access_token(
    form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/login/token")  #new

#new function, It works as a dependency
def get_current_user_from_token(token: str = Depends(oauth2_scheme),db: Session=Depends(get_db)): 
    credentials_exception = HTTPException(
        detail="Could not validate credentials",
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        username: str = payload.get("sub")
        print("username/email extracted is ",username)
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = get_user(username=username,db=db)
    if user is None:
        raise credentials_exception
    return user

Lets integrate permissions with our post view to attach real user id instead of hardcoded user id. And we are also going to attach this dependency with our route for delete, So, that only the users who have created the post or is superuser then only can delete the job post.

from typing import List
from sqlalchemy.orm import Session
from db.models.users import User #new
from apis.version1.route_login import get_current_user_from_token  #new

router = APIRouter()

@router.post("/create-job/", response_model=ShowJob)
def create_job(job: JobCreate, db: Session = Depends(get_db),current_user:User = Depends(get_current_user_from_token)):  #new dependency here
    job = create_new_job(job=job, db=db, owner_id=current_user.id)
    return job

def delete_job(id: int,db: Session = Depends(get_db),current_user: User = Depends(get_current_user_from_token)):
    job = retreive_job(id =id,db=db)
    if not job:
        return HTTPException(status_code=status.HTTP_404_NOT_FOUND,detail=f"Job with {id} does not exist")
    if job.owner_id == current_user.id or current_user.is_superuser:
        return {"msg":"Successfully deleted."}
    raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
                            detail=f"You are not permitted!!!!")

All done, lets give it a try.🙈

Final git commit : add support for authorization · nofoobar/JobBoard-Fastapi@12b168a (github.com)

Prev: 23 : Authentication … Next: 25 : Unit …

Brige the gap between Tutorial hell and Industry. We want to bring in the culture of Clean Code, Test Driven Development.

We know, we might make it hard for you but definitely worth the efforts.



Refund Policy

Follow us on our social media channels to stay updated.

© Copyright 2022-23 Team FastAPITutorial