deepvasoya/health-apps-cms
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: staff management apis

Browse Source
This commit is contained in:
deepvasoya
2025-05-21 15:59:33 +05:30
parent 7cae614e37
commit 9c39d369c1
13 changed files with 360 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/authService.py
+123 -4
View File
@@ -1,17 +1,28 @@
from operator import or_
import os
import dotenv
dotenv.load_dotenv()
from interface.common_response import CommonResponse
from schemas.ResponseSchemas import UserResponse
import datetime
import json
import urllib.request
from sqlalchemy.orm import Session
from services.jwtService import create_jwt_token
from services.userServices import UserServices
from models import BlockedEmail
from services.emailService import EmailService
from exceptions.validation_exception import ValidationException
from models import OTP
from enums.enums import UserType
from models import Users
from exceptions.resource_not_found_exception import ResourceNotFoundException
from models import ResetPasswordTokens
from utils.constants import generateOTP
from utils.password_utils import verify_password
from utils.password_utils import generate_reset_password_token, generate_secure_password, hash_password, verify_password
from schemas.CreateSchemas import UserCreate
from schemas.BaseSchemas import AuthBase, AuthOTP
from schemas.BaseSchemas import AuthBase, AuthOTP, CreateSuperAdmin
from exceptions.unauthorized_exception import UnauthorizedException
from database import get_db
@@ -23,6 +34,7 @@ class AuthService:
self.user_service = UserServices()
self.db = next(get_db())
self.email_service = EmailService()
self.url = os.getenv("FRONTEND_URL")
def login(self, data: AuthBase) -> str:
@@ -106,4 +118,111 @@ class AuthService:
# self.db.delete(db_otp)
# self.db.commit()
return
return
def get_admins(self, user, limit:int, offset:int, search:str):
try:
if user["userType"] != UserType.SUPER_ADMIN:
raise UnauthorizedException("User is not authorized to perform this action")
admins = self.db.query(Users).filter(Users.userType == UserType.SUPER_ADMIN)
total = self.db.query(Users).filter(Users.userType == UserType.SUPER_ADMIN).count()
if search:
admins = admins.filter(
or_(
Users.username.contains(search),
Users.email.contains(search),
)
)
total = admins.count()
admins = admins.limit(limit).offset(offset).all()
response = [UserResponse(**admin.__dict__.copy()) for admin in admins]
common_response = CommonResponse(data=response, total=total)
return common_response
except Exception as e:
raise e
def create_super_admin(self, user, data: CreateSuperAdmin):
if user["userType"] != UserType.SUPER_ADMIN:
raise UnauthorizedException("User is not authorized to perform this action")
password = generate_secure_password()
hashed_password = hash_password(password)
# check if username and email are unique
existing_user = (
self.db.query(Users)
.filter(
Users.email == data.email.lower(),
)
.first()
)
if existing_user:
raise ValidationException("User with same email already exists")
user = Users(
username=data.username.lower(),
email=data.email.lower(),
password=hashed_password,
userType=UserType.SUPER_ADMIN,
)
self.db.add(user)
self.db.commit()
LOGIN_URL = self.url
# send email to user
self.email_service.send_new_admin_email(data.email, password, LOGIN_URL)
return
def forget_password(self, email: str):
user = self.db.query(Users).filter(Users.email == email.lower()).first()
if not user:
raise ResourceNotFoundException("User not found")
# get reset password token
reset_password_token = generate_reset_password_token()
reset_password = ResetPasswordTokens(email=email, token=reset_password_token)
self.db.add(reset_password)
self.db.commit()
reset_password_url = f"{self.url}/auth/reset-password?token={reset_password_token}"
self.email_service.send_reset_password_email(email, reset_password_url)
return
def reset_password(self, token: str, password: str):
reset_password = self.db.query(ResetPasswordTokens).filter(ResetPasswordTokens.token == token).first()
if not reset_password:
raise ResourceNotFoundException("Reset password token not found")
user = self.db.query(Users).filter(Users.email == reset_password.email).first()
if not user:
raise ResourceNotFoundException("User not found")
user.password = hash_password(password)
self.db.delete(reset_password)
self.db.commit()
return
services/emailService.py
+57 -4
View File
@@ -67,10 +67,45 @@ class EmailService:
"TextPart": "Dear User, Congratulations! Your clinic {{name}} has been approved. Thank you, Team 24x7 AIHR"
}
self.client.create_template(Template=otp_template)
self.client.create_template(Template=new_clinic_template)
self.client.create_template(Template=reject_clinic_template)
self.client.create_template(Template=apporve_clinic_template)
new_admin_template = {
"TemplateName": "newAdmin",
"SubjectPart": "Login Credentials",
"HtmlPart": """
<p>Dear User,</p>
<p>Your login credentials are:</p>
<div>
<p>Email: {{email}}</p>
<p>Password: {{password}}</p>
</div>
<br />
<p>Use the following link to login:</p>
<p>Login URL: {{login_url}}</p>
<br />
<p>Thank you,<br/>Team 24x7 AIHR</p>
""",
"TextPart": "Dear User, Your login credentials are: Email: {{email}} Password: {{password}} Login URL: {{login_url}} Thank you, Team 24x7 AIHR"
}
reset_password_template = {
"TemplateName": "resetPassword",
"SubjectPart": "Reset Password",
"HtmlPart": """
<p>Dear User,</p>
<p>You have requested to reset your password. Please use the following link to reset your password:</p>
<p>Reset Password URL: {{reset_password_url}}</p>
<br />
<p>Thank you,<br/>Team 24x7 AIHR</p>
""",
"TextPart": "Dear User, You have requested to reset your password. Please use the following link to reset your password: Reset Password URL: {{reset_password_url}} Thank you, Team 24x7 AIHR"
}
# self.client.create_template(Template=otp_template)
# self.client.create_template(Template=new_clinic_template)
# self.client.create_template(Template=reject_clinic_template)
# self.client.create_template(Template=apporve_clinic_template)
# self.client.create_template(Template=new_admin_template)
# self.client.create_template(Template=new_admin_template)
self.client.create_template(Template=reset_password_template)
except Exception as e:
logger.error(f"Failed to create template: {e}")
@@ -132,4 +167,22 @@ class EmailService:
to_address=email,
template_data={"clinic_name": clinic_name, "email": email}
)
return
def send_new_admin_email(self, email: str, password:str, login_url:str):
"""Send new admin email"""
self.send_email(
template_name="newAdmin",
to_address=email,
template_data={"email": email, "password": password, "login_url": login_url}
)
return
def send_reset_password_email(self, email: str, reset_password_url: str):
"""Send reset password email"""
self.send_email(
template_name="resetPassword",
to_address=email,
template_data={"reset_password_url": reset_password_url}
)
return
services/userServices.py
+10 -7
View File
@@ -186,17 +186,20 @@ class UserServices:
return response
def get_user_by_email(self, email: str) -> UserResponse:
user = self.db.query(Users).filter(Users.email == email.lower()).first()
try:
user = self.db.query(Users).filter(Users.email == email.lower()).first()
if not user:
logger.error("User not found")
raise ResourceNotFoundException("User not found")
if not user:
logger.error("User not found")
raise ResourceNotFoundException("User not found")
user_dict = user.__dict__.copy()
user_dict = user.__dict__.copy()
user_response = UserResponse(**user_dict)
user_response = UserResponse(**user_dict)
return user_response
return user_response
except Exception as e:
raise e
def update_user(self, admin_id:int|None, user_id: int, user_data: UserUpdate) -> UserResponse:
# Check admin authorization if admin_id is provided