Files
full-stack-fastapi-postgresql/{{cookiecutter.project_slug}}/backend/app/app/api/utils/security.py
Stephen Brown II 79631c7619 Use Pydantic BaseSettings for config settings (#87)
* Use Pydantic BaseSettings for config settings

* Update fastapi dep to >=0.47.0 and email_validator to email-validator

* Fix deprecation warning for Pydantic >=1.0

* Properly support old-format comma separated strings for BACKEND_CORS_ORIGINS

Co-authored-by: Sebastián Ramírez <tiangolo@gmail.com>
2020-04-17 07:56:10 +02:00

46 lines
1.5 KiB
Python

import jwt
from fastapi import Depends, HTTPException, Security
from fastapi.security import OAuth2PasswordBearer
from jwt import PyJWTError
from sqlalchemy.orm import Session
from starlette.status import HTTP_403_FORBIDDEN
from app import crud
from app.api.utils.db import get_db
from app.core.config import settings
from app.core.jwt import ALGORITHM
from app.models.user import User
from app.schemas.token import TokenPayload
reusable_oauth2 = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/login/access-token")
def get_current_user(
db: Session = Depends(get_db), token: str = Security(reusable_oauth2)
):
try:
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[ALGORITHM])
token_data = TokenPayload(**payload)
except PyJWTError:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Could not validate credentials"
)
user = crud.user.get(db, id=token_data.user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
def get_current_active_user(current_user: User = Security(get_current_user)):
if not crud.user.is_active(current_user):
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
def get_current_active_superuser(current_user: User = Security(get_current_user)):
if not crud.user.is_superuser(current_user):
raise HTTPException(
status_code=400, detail="The user doesn't have enough privileges"
)
return current_user