-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathauthController.py
More file actions
46 lines (35 loc) · 1.87 KB
/
Copy pathauthController.py
File metadata and controls
46 lines (35 loc) · 1.87 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
from datetime import timedelta
from fastapi import APIRouter, HTTPException, Header
from src.services.userService import UserService
from ._helpers import user_dependency, form_auth_dependency
from src.infra.security.hashing import ALGORITHM, create_token, ACCESS_TOKEN_EXPIRE_MINUTES, REFRESH_TOKEN_EXPIRE_DAYS, JWT_ACCESS_SECRET_KEY, JWT_REFRESH_SECRET_KEY, bcrypt_context, decode_token
router = APIRouter(prefix="/auth", tags=["auth"])
@router.post("/login")
def login(form_data: form_auth_dependency):
email = form_data.username.lower()
try:
user = UserService().view_by_email(email)
except Exception:
raise HTTPException(status_code=404, detail={"message": "Email or password incorrect.", "error": True})
if not bcrypt_context.verify(form_data.password, user["senha"]):
raise HTTPException(status_code=401, detail={"message": "Email or password incorrect.", "error": True})
access_token = create_token(user["id"], JWT_ACCESS_SECRET_KEY, timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
refresh_token = create_token(user["id"], JWT_REFRESH_SECRET_KEY, timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS))
return {
"access_token": access_token,
"refresh_token": refresh_token,
"token_type": "Bearer"
}
@router.post("/refresh")
def refresh(refresh_token: str = Header(..., alias="X-Refresh-Token")):
payload = decode_token(refresh_token, JWT_REFRESH_SECRET_KEY, [ALGORITHM])
user_id = int(payload["sub"])
try:
UserService().view(user_id)
except Exception:
raise HTTPException(status_code=401, detail={"message": "User not found", "error": True})
new_access_token = create_token(payload["sub"], JWT_ACCESS_SECRET_KEY, timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
return {"access_token": new_access_token, "error": False}
@router.get("/me")
def me(user: user_dependency):
return user