user process

auth/models.py

@@ -1,4 +1,6 @@
 from enum import Enum
+import random
+import smtplib
 from backend.config import SECRET_KEY, ALGORITHM, ACCESS_TOKEN_EXPIRE_MINUTES ,pwd_context, get_session_db, Base
 from datetime import datetime, timedelta, timezone
 from pydantic import BaseModel
@@ -8,6 +10,7 @@ from fastapi.security import OAuth2PasswordBearer
 from pydantic.networks import EmailStr
 from sqlalchemy import Column, Integer, String
 from sqlalchemy.orm import Session
+from email.message import EmailMessage
 import jwt
 
 class Token(BaseModel):
@@ -171,3 +174,67 @@ def register_user(
     session.commit()
     session.refresh(db_user)
     return db_user
+
+
+def find_user_w_email(
+        session: Annotated[Session, Depends(get_session_db)],
+        email: EmailStr | None = None,
+):
+    
+    exist_user  = session.query(DBUser).filter(DBUser.email == email).first() #email ile kullanıcıyı bul
+    if exist_user is None:
+        raise HTTPException(status_code=400, detail="User not found")
+    
+    if exist_user.status == Status.banned:
+        raise HTTPException(status_code=400, detail="Inactive user")
+    
+    return True
+    
+def send_password_to_email(
+        session: Annotated[Session, Depends(get_session_db)],
+        email: EmailStr | None = None,
+) -> str:
+    
+    msg = EmailMessage() #obje oluştur
+    msg['Subject'] = 'Password Reset'
+    msg['From'] = 'hansneiumann@gmail.com'
+    msg['To'] = email
+    veritification_code = generate_password_reset_number()
+    msg.set_content(veritification_code)
+
+    with smtplib.SMTP_SSL('smtp.gmail.com', 465) as smtp:
+        smtp.login("hansneiumann@gmail.com", "rwaq mbil lzut dgja")
+        smtp.send_message(msg)
+
+    update_password_w_email(session, email=email, password=veritification_code) #şifreyi güncelle
+
+
+def generate_password_reset_number() -> str:
+    return str(random.randint(10000000, 99999999)) # 8 haneli rastgele bir sayı döndür
+
+
+def update_password_w_email(
+        session: Annotated[Session, Depends(get_session_db)],
+        password: str | None = None,
+        email: EmailStr | None = None,
+) -> dict:
+        
+    hashed_password = get_password_hash(password)
+    session.query(DBUser).filter(DBUser.email == email).update({"hashed_password": hashed_password})
+    session.commit()
+
+    return {"message": "Password updated successfully"}
+
+def update_password_w_user(
+        session: Annotated[Session, Depends(get_session_db)],
+        user: Annotated[DBUser , None],
+        password: str | None = None,
+        
+):
+    
+    hashed_password = get_password_hash(password)
+    session.query(DBUser).filter(DBUser.user_id == user.user_id).update({"hashed_password": hashed_password})
+    session.commit()
+    session.refresh(user)
+    
+    

auth/router.py

@@ -1,10 +1,12 @@
-from fastapi import APIRouter, Depends, HTTPException, status
-from .models import Token, UserPublic, authenticate_user, create_access_token, UserCreate, register_user
+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
+
 
 router = APIRouter(
     prefix="/auth",
@@ -41,4 +43,33 @@ async def create_user(
     
     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."}
+    
+