51 lines
1.7 KiB
Python
51 lines
1.7 KiB
Python
from fastapi import APIRouter, Depends, status, HTTPException, Response
|
|
from fastapi.security.oauth2 import OAuth2PasswordRequestForm
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .. import database, schemas, models, security, utils
|
|
|
|
router = APIRouter(tags=['Authentication'])
|
|
|
|
|
|
@router.post('/login', response_model=schemas.Token)
|
|
def login(user_credentials: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(database.get_db)):
|
|
|
|
user = db.query(models.Users).filter(
|
|
models.Users.Email == user_credentials.username).first()
|
|
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN, detail="Invalid Credentials")
|
|
|
|
if not utils.verify(user_credentials.password, user.Hashed_password):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN, detail="Invalid Credentials")
|
|
|
|
access_token = security.create_access_token(data={
|
|
"UserID": user.UserID,
|
|
"Email": user.Email
|
|
})
|
|
|
|
refresh_token = security.create_refresh_token(data={
|
|
"UserID": user.UserID,
|
|
})
|
|
|
|
return {"access_token": access_token, "refresh_token": refresh_token}
|
|
|
|
@router.post('/refresh', response_model=schemas.Token, response_model_exclude_none=True )
|
|
def refresh_access_token(refresh_token: str, db: Session = Depends(database.get_db)):
|
|
user_id = security.verify_refresh_token(refresh_token)
|
|
|
|
user = db.query(models.Users).filter(
|
|
models.Users.UserID == user_id).first()
|
|
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
|
|
|
|
access_token = security.create_access_token(data={
|
|
"UserID": user.UserID,
|
|
"Email": user.Email
|
|
})
|
|
|
|
return {"access_token": access_token} |