75 lines
2.6 KiB
Python
75 lines
2.6 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status, BackgroundTasks
|
||
from .models import Token, UserPublic, authenticate_user, create_access_token, UserCreate, find_user_w_email, get_current_user, register_user, send_password_to_email, update_password_w_user
|
||
from datetime import timedelta
|
||
from typing import Annotated
|
||
from ..config import get_session_db
|
||
from fastapi.security import OAuth2PasswordRequestForm
|
||
from sqlalchemy.orm import Session
|
||
from pydantic.networks import EmailStr
|
||
from ..tasks import appCelery
|
||
|
||
|
||
router = APIRouter(
|
||
prefix="/auth",
|
||
tags=["auth"],
|
||
responses={404: {"description": "Not found"}},
|
||
dependencies=[],
|
||
)
|
||
|
||
@router.post('/login')
|
||
async def login_for_access_token(
|
||
form_data : Annotated[OAuth2PasswordRequestForm, Depends()],
|
||
session : Annotated[Session, Depends(get_session_db)],
|
||
) -> Token:
|
||
|
||
user = authenticate_user(session, form_data.username, form_data.password)
|
||
if not user:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="Incorrect username or password",
|
||
headers={"WWW-Authenticate": "Bearer"},
|
||
)
|
||
access_token_expires = timedelta(minutes=30)
|
||
access_token = create_access_token(
|
||
data={"sub": user.username, "role": user.role, 'status': user.status}, expires_delta=access_token_expires
|
||
)
|
||
return Token(access_token=access_token, token_type="bearer")
|
||
|
||
|
||
@router.post('/register', response_model=UserPublic) #userPublic güvenli bir model
|
||
async def create_user(
|
||
session : Annotated[Session, Depends(get_session_db)],
|
||
user : Annotated[UserCreate, Depends()]
|
||
):
|
||
|
||
return register_user(session, user)
|
||
|
||
@router.post('/password_reset')
|
||
async def password_reset(
|
||
session : Annotated[Session, Depends(get_session_db)],
|
||
email : Annotated[EmailStr, None] = None,
|
||
task: Annotated[BackgroundTasks, None] = None, # BackgroundTasks, task'ı arka planda çalıştırmak için kullanıyoruz
|
||
):
|
||
|
||
if not find_user_w_email(session, email):
|
||
return HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail="bad request",
|
||
)
|
||
|
||
task.add_task(send_password_to_email, session, email)
|
||
return {"message": "New password has been sent to your email."}
|
||
|
||
@router.post('/update_password')
|
||
async def update_password(
|
||
user: Annotated[str, Depends(get_current_user)],
|
||
session: Annotated[Session, Depends(get_session_db)],
|
||
new_password: Annotated[str, None] = None,
|
||
) -> dict:
|
||
|
||
update_password_w_user(session, user, new_password)
|
||
|
||
return {"message": "Password updated successfully."}
|
||
|
||
|
||
|