fastapi
pip install fastapi
pip install uvicorn # ASGI server
pip install starlette # lightweight ASGI framework/toolkit
pip install pydantic # Data validation and type annotations
# OR
pip install fastapi uvicorn starlette pydanticfastapi
pip install fastapi
pip install uvicorn # ASGI server
pip install starlette # lightweight ASGI framework/toolkit
pip install pydantic # Data validation and type annotations
# OR
pip install fastapi uvicorn starlette pydanticfastapi authentication
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='token')
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
async def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
id: int = payload.get("id")
if id is None:
raise credentials_exception
except JWTError:
raise credentials_exception
user = db.get(User, id)
if user is None:
raise credentials_exception
return user
async def check_if_admin(user: schemas.User = Depends(get_current_user)):
if user.role == "admin":
return user
else:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="User is not admin",
headers={"WWW-Authenticate": "Bearer"},
)
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 authenticate_user(db: Session, username: str, password: str):
try:
user: User = db.execute(select(User).filter(User.username == username).filter(User.password == password)).scalar_one()
data_dict = {"id": user.user_id, "username": user.username, "role": user.role}
return create_access_token(data_dict)
except Exception as e:
print(e)
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
@app.post('/token')
async def token(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
token = authenticate_user(db, form_data.username, form_data.password)
return {"access_token": token, "token_type": "bearer"}
@app.post("/admin/example_admin_route", dependencies=[Depends(check_if_admin)])
def example_admin_route():Copyright © 2021 Codeinu
Forgot your account's password or having trouble logging into your Account? Don't worry, we'll help you to get back your account. Enter your email address and we'll send you a recovery link to reset your password. If you are experiencing problems resetting your password contact us
