Password Reset Email Link

api/utils/auth.py

@@ -12,6 +12,8 @@
 from api.v1.schemas.token import TokenData
 from api.db.database import get_db
 from .config import SECRET_KEY, ALGORITHM
+from urllib.parse import urlencode
+import uuid
 
 # Initialize OAuth2PasswordBearer
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")
@@ -27,6 +29,7 @@ def verify_password(plain_password: str, hashed_password: str) -> bool:
 def get_user(db, username: str):
     return db.query(User).filter(User.username == username).first()
 
+
 def authenticate_user(db, username: str, password: str):
     user = get_user(db, username)
     if not user:
@@ -43,4 +46,21 @@ def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
         expire = datetime.utcnow() + timedelta(minutes=15)
     to_encode.update({"exp": expire})
     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
-    return encoded_jwt
+    return encoded_jwt
+
+
+def get_user_by_email(db: Session, email: str):
+    return db.query(User).filter(User.email == email).first()
+
+
+def generate_password_reset_token(user_id: uuid.UUID) -> str:
+    expires_delta = timedelta(minutes=30)
+    data = {"sub": user_id}
+    return create_access_token(data=data, expires_delta=expires_delta)
+
+
+def generate_reset_password_url(user_id: uuid.UUID, token: str) -> str:
+    base_url = f"http://localhost:7001"
+    path = "/reset-password"
+    query_params = urlencode({"token": token, "user_id": user_id})
+    return f"{base_url}{path}?{query_params}"

api/utils/settings.py

@@ -28,6 +28,8 @@ class Settings(BaseSettings):
     MAIL_FROM: str = config('MAIL_FROM')
     MAIL_PORT: int = config('MAIL_PORT')
     MAIL_SERVER: str = config('MAIL_SERVER')
+    MAIL_STARTTLS: bool = config('MAIL_TLS')
+    MAIL_SSL_TLS: bool = config('MAIL_SSL')
 
 
 settings = Settings()

api/v1/models/user.py

@@ -21,11 +21,13 @@
 from api.v1.models.base import Base, user_organization_association, user_role_association
 from api.v1.models.base_model import BaseModel
 from sqlalchemy.dialects.postgresql import UUID
+import uuid
 
 
 class User(BaseModel, Base):
     __tablename__ = 'users'
 
+    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
     username = Column(String(50), unique=True, nullable=False)
     email = Column(String(100), unique=True, nullable=False)
     password = Column(String(255), nullable=False)

api/v1/routes/auth.py

@@ -18,20 +18,35 @@
 from api.v1.schemas.token import Token, LoginRequest
 from api.v1.schemas.auth import UserBase, SuccessResponse, SuccessResponseData, UserCreate
 from api.db.database import get_db
-from api.utils.auth import authenticate_user, create_access_token,hash_password,get_user
+from api.utils.auth import (
+    authenticate_user,
+    create_access_token,
+    hash_password,get_user,
+    generate_password_reset_token,
+    generate_reset_password_url,
+    get_user_by_email
+)
 from api.utils.dependencies import get_current_admin, get_current_user
 
 
 from api.v1.models.org import Organization
 
 from api.v1.models.product import Product
 
+from fastapi import BackgroundTasks
+from fastapi_mail import FastMail, MessageSchema
+
+from fastapi import BackgroundTasks
+from fastapi_mail import ConnectionConfig
+from pydantic import BaseModel, EmailStr
+from api.utils.settings import settings, BASE_DIR
 
 db = next(get_db())
 
 
 auth = APIRouter(prefix="/auth", tags=["auth"])
 
+# This is in the .evn file already though
 ACCESS_TOKEN_EXPIRE_MINUTES = 30
 
 
@@ -111,3 +126,51 @@ def register_user(user: UserCreate, db: Session = Depends(get_db)):
 def read_admin_data(current_admin: Annotated[User, Depends(get_current_admin)]):
     return {"message": "Hello, admin!"}
 
+
+conf = ConnectionConfig(
+    MAIL_USERNAME=settings.MAIL_USERNAME,
+    MAIL_PASSWORD=settings.MAIL_PASSWORD,
+    MAIL_FROM=settings.MAIL_FROM,
+    MAIL_PORT=settings.MAIL_PORT,
+    MAIL_SERVER=settings.MAIL_SERVER,
+    MAIL_STARTTLS=settings.MAIL_STARTTLS,
+    MAIL_SSL_TLS=settings.MAIL_SSL_TLS
+)
+
+
+class EmailRequest(BaseModel):
+    email: EmailStr
+
+
+@auth.post("/password-reset-email/", status_code=status.HTTP_200_OK)
+async def send_email(background_tasks: BackgroundTasks, email: EmailRequest, db: Session = Depends(get_db)):
+    """
+    Getting the user from the database, the email in the db since the email schema in the db is unique, it picks the 1st
+    """
+
+    user = get_user_by_email(db, email.email)
+    if not user:
+        raise HTTPException(status_code=404, detail="We don't have user with the provided email in our database.")
+    # Generate password reset token
+    password_reset_token = generate_password_reset_token(user_id=str(user.id))
+    reset_password_url = generate_reset_password_url(user_id=str(user.id), token=password_reset_token)
+
+    email_body = (f"Dear {user.username}!\nYou requested for email reset on our site.\n"
+                  f"To reset your password, click the following link: {reset_password_url}")
+
+    message: MessageSchema = MessageSchema(
+        subject="Reset Password",
+        recipients=[email.email],
+        body=email_body,
+        subtype="plain"
+    )
+    fm = FastMail(conf)
+
+    try:
+        background_tasks.add_task(fm.send_message, message)
+        return {
+            "message": "Password reset email sent successfully.",
+            "reset_link": reset_password_url
+        }
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Error sending email: {e}")