Feature to deactivate a user

.env.sample

@@ -13,3 +13,9 @@ ALGORITHM = HS256
 ACCESS_TOKEN_EXPIRE_MINUTES = 10
 JWT_REFRESH_EXPIRY=5
 APP_URL=
+
+MAIL_USERNAME=""
+MAIL_PASSWORD=""
+MAIL_FROM=""
+MAIL_PORT=465
+MAIL_SERVER="smtp.gmail.com"

api/core/dependencies/email.py

@@ -0,0 +1,28 @@
+import smtplib
+from typing import List, Optional, Union
+
+from fastapi import HTTPException
+from api.utils.settings import settings
+
+
+class MailService:
+    '''Class to send different emails for different services'''
+
+    def send_mail(self, to: str, subject: str, body: str):
+        '''Function to send email to a user either as a regular test or as html file'''
+
+        try:
+            with smtplib.SMTP(settings.MAIL_SERVER, settings.MAIL_PORT) as conn:
+                conn.starttls()
+                conn.login(user=settings.MAIL_USERNAME, password=settings.MAIL_PASSWORD) 
+                conn.sendmail(
+                    from_addr=settings.MAIL_FROM,
+                    to_addrs=to,
+                    msg=f"Subject:{subject}\n\n{body}"
+                )
+
+        except smtplib.SMTPException as smtp_error:
+            raise HTTPException(500, f'SMTP ERROR- {smtp_error}')
+
+
+mail_service = MailService()

api/utils/settings.py

@@ -23,5 +23,11 @@ class Settings(BaseSettings):
     DB_NAME: str = config("DB_NAME")
     DB_TYPE: str = config("DB_TYPE")
 
+    MAIL_USERNAME: str = config("MAIL_USERNAME")
+    MAIL_PASSWORD: str = config('MAIL_PASSWORD')
+    MAIL_FROM: str = config('MAIL_FROM')
+    MAIL_PORT: int = config('MAIL_PORT')
+    MAIL_SERVER: str = config('MAIL_SERVER')
+
 
 settings = Settings()

api/v1/routes/user.py

@@ -0,0 +1,81 @@
+from fastapi import Depends, HTTPException, APIRouter, Request
+from jose import JWTError
+import jwt
+from sqlalchemy.orm import Session
+
+from api.core.dependencies.email import mail_service
+from api.utils.auth import create_access_token
+
+from api.utils.config import SECRET_KEY, ALGORITHM
+from ..models.user import User
+from api.v1.schemas.user import DeactivateUserSchema
+from api.db.database import get_db
+from api.utils.dependencies import get_current_user
+
+
+user = APIRouter(prefix='/api/v1/users', tags=['Users'])
+
+@user.patch('/accounts/deactivate', status_code=200)
+async def deactivate_account(request: Request, schema: DeactivateUserSchema, db: Session = Depends(get_db), user: User = Depends(get_current_user)):
+    '''Endpoint to deactivate a user account'''
+
+    # Generate an access token containing user credentials
+    token = create_access_token(data={'username': f'{user.username}'})
+
+    if not user.is_active:
+        raise HTTPException(400, 'User is inactive')
+
+    if schema.confirmation == False:
+        raise HTTPException(400, 'Confirmation required to deactivate account')
+
+    user.is_active = False
+
+    # Send aail to user
+    mail_service.send_mail(
+        to=user.email, 
+        subject='Account deactivation', 
+        body=f'Hello, {user.first_name},\n\nYour account has been deactivated successfully.\nTo reactivate your account if this was a mistake, please click the link below:\n{request.url.hostname}/api/users/accounts/reactivate?token={token}\n\nThis link expires after 15 minutes.'
+    )
+
+    # Commit changes to deactivate the user
+    db.commit()
+
+    return {"status_code": 200, "message": "Account deactivated successfully. Check email for confirmation"}
+
+
+@user.get('/accounts/reactivate', status_code=200)
+async def reactivate_account(request: Request, db: Session = Depends(get_db)):
+    '''Endpoint to reactivate a user account'''
+
+    # Get access token from query
+    token = request.query_params.get('token')
+
+    # Validate the token
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        username = payload.get('username')
+
+        if username is None:
+            raise HTTPException(400, 'Invalid token')
+
+    except JWTError:
+        raise HTTPException(400, 'Invalid token')
+
+    user = db.query(User).filter(User.username == username).first()
+
+    if user.is_active:
+        raise HTTPException(400, 'User is already active')
+
+    user.is_active = True
+
+    # Send aail to user
+    mail_service.send_mail(
+        to=user.email, 
+        subject='Account reactivation', 
+        body=f'Hello, {user.first_name},\n\nYour account has been reactivated successfully'
+    )
+
+    # Commit changes to deactivate the user
+    db.commit()
+
+    return {"status_code": 200, "message": "Account reactivated successfully. Check email for confirmation"}

api/v1/schemas/user.py

@@ -0,0 +1,10 @@
+from typing import Optional
+from pydantic import BaseModel
+
+
+class DeactivateUserSchema(BaseModel):
+    '''Schema for deactivating a user'''
+
+    reason: Optional[str] = None
+    confirmation: bool
+

main.py

@@ -6,12 +6,15 @@
 from starlette.requests import Request
 from api.db.database import Base, engine
 
+from api.v1.routes.newsletter_router import newsletter
 from api.v1.routes.newsletter_router import (
     CustomException,
     custom_exception_handler
 )
 
-from api.v1.routes import api_version_one
+from api.v1.routes.auth import auth
+from api.v1.routes.user import user
+from api.v1.routes.roles import role
 
 Base.metadata.create_all(bind=engine)
 
@@ -35,9 +38,11 @@ async def lifespan(app: FastAPI):
 )
 
 app.add_exception_handler(CustomException, custom_exception_handler) # Newsletter custom exception registration
+app.include_router(newsletter, tags=["Newsletter"])
 
-app.include_router(api_version_one)
-
+app.include_router(auth)
+app.include_router(user)
+# app.include_router(users, tags=["Users"])
 
 
 @app.get("/", tags=["Home"])

tests/database.py

@@ -48,4 +48,5 @@ def override_get_db():
         finally:
             session.close()
     app.dependency_overrides[get_db] = override_get_db
-    yield TestClient(app)
+    yield TestClient(app)
+