fastapi authentication

1pip install fastapi
2pip install uvicorn # ASGI server
3pip install starlette # lightweight ASGI framework/toolkit
4pip install pydantic # Data validation and type annotations
5# OR
6pip install fastapi uvicorn starlette pydantic

1oauth2_scheme = OAuth2PasswordBearer(tokenUrl='token')
2
3def get_db():
4    db = SessionLocal()
5    try:
6        yield db
7    finally:
8        db.close()
9
10
11async def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
12    credentials_exception = HTTPException(
13        status_code=status.HTTP_401_UNAUTHORIZED,
14        detail="Could not validate credentials",
15        headers={"WWW-Authenticate": "Bearer"},
16    )
17    try:
18        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
19        id: int = payload.get("id")
20        if id is None:
21            raise credentials_exception
22    except JWTError:
23        raise credentials_exception
24    user = db.get(User, id)
25    if user is None:
26        raise credentials_exception
27    return user
28
29
30async def check_if_admin(user: schemas.User = Depends(get_current_user)):
31    if user.role == "admin":
32        return user
33    else:
34        raise HTTPException(
35            status_code=status.HTTP_401_UNAUTHORIZED,
36            detail="User is not admin",
37            headers={"WWW-Authenticate": "Bearer"},
38        )
39
40
41def create_access_token(data: dict):
42    to_encode = data.copy()
43    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
44    to_encode.update({"exp": expire})
45    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
46    return encoded_jwt
47
48
49def authenticate_user(db: Session, username: str, password: str):
50    try:
51        user: User = db.execute(select(User).filter(User.username == username).filter(User.password == password)).scalar_one()
52        data_dict = {"id": user.user_id, "username": user.username, "role": user.role}
53        return create_access_token(data_dict)
54    except Exception as e:
55        print(e)
56        raise HTTPException(
57            status_code=status.HTTP_401_UNAUTHORIZED,
58            detail="Incorrect username or password",
59            headers={"WWW-Authenticate": "Bearer"},
60        )
61@app.post('/token')
62async def token(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
63    token = authenticate_user(db, form_data.username, form_data.password)
64    return {"access_token": token, "token_type": "bearer"}
65    
66@app.post("/admin/example_admin_route", dependencies=[Depends(check_if_admin)])
67def example_admin_route():