124 lines
4.8 KiB
Python
124 lines
4.8 KiB
Python
from sqlalchemy.orm import Session
|
|
from fastapi import HTTPException, status
|
|
from app.repositories.user import user_repository
|
|
from app.schemas.user import UserCreate, UserUpdate, Token
|
|
from app.utils.security import get_password_hash, verify_password, create_access_token
|
|
from typing import Optional, List
|
|
|
|
class AuthService:
|
|
"""
|
|
Service containing authentication-related business logic.
|
|
Bridge between the controllers and repositories.
|
|
"""
|
|
@staticmethod
|
|
def register(db: Session, user_in: UserCreate):
|
|
# 1. Check if user already exists by username
|
|
existing_user = user_repository.get_by_username(db, user_in.username)
|
|
if existing_user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Username sudah terdaftar"
|
|
)
|
|
|
|
# 2. Check if user already exists by email
|
|
existing_email = user_repository.get_by_email(db, user_in.email)
|
|
if existing_email:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Email sudah terdaftar"
|
|
)
|
|
|
|
# 3. Hash the user password
|
|
hashed_password = get_password_hash(user_in.password)
|
|
|
|
# 4. Save user to database
|
|
db_user_data = {
|
|
"username": user_in.username,
|
|
"nama": user_in.nama or user_in.username,
|
|
"email": user_in.email,
|
|
"role": user_in.role or "admin",
|
|
"kode_prodi": user_in.kode_prodi,
|
|
"hashed_password": hashed_password,
|
|
"is_active": True
|
|
}
|
|
return user_repository.create(db, db_user_data)
|
|
|
|
@staticmethod
|
|
def login(db: Session, username: str, password: str) -> Token:
|
|
# 1. Retrieve user by username
|
|
user = user_repository.get_by_username(db, username)
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Username atau password salah",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
# 2. Verify password hash matches plain text
|
|
if not verify_password(password, user.hashed_password):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Username atau password salah",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
# 3. Ensure user is active
|
|
if not user.is_active:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Akun dinonaktifkan. Silakan hubungi Administrator."
|
|
)
|
|
|
|
# 4. Generate JWT token
|
|
access_token = create_access_token(subject=user.username)
|
|
return Token(
|
|
access_token=access_token,
|
|
token_type="bearer",
|
|
username=user.username,
|
|
nama=user.nama or user.username,
|
|
role=user.role or "admin",
|
|
kode_prodi=user.kode_prodi
|
|
)
|
|
|
|
@staticmethod
|
|
def get_all_users(db: Session):
|
|
return user_repository.get_all(db, skip=0, limit=200)
|
|
|
|
@staticmethod
|
|
def update_user(db: Session, user_id: int, user_in: UserUpdate):
|
|
user = user_repository.get(db, user_id)
|
|
if not user:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Pengguna tidak ditemukan")
|
|
|
|
# Check email duplicate
|
|
if user_in.email and user_in.email != user.email:
|
|
existing_email = user_repository.get_by_email(db, user_in.email)
|
|
if existing_email and existing_email.id != user_id:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Email sudah digunakan oleh akun lain")
|
|
|
|
update_data = {}
|
|
if user_in.nama is not None:
|
|
update_data["nama"] = user_in.nama
|
|
if user_in.email is not None:
|
|
update_data["email"] = user_in.email
|
|
if user_in.role is not None:
|
|
update_data["role"] = user_in.role
|
|
if user_in.kode_prodi is not None:
|
|
update_data["kode_prodi"] = user_in.kode_prodi
|
|
if user_in.is_active is not None:
|
|
update_data["is_active"] = user_in.is_active
|
|
if user_in.password:
|
|
update_data["hashed_password"] = get_password_hash(user_in.password)
|
|
|
|
return user_repository.update(db, user, update_data)
|
|
|
|
@staticmethod
|
|
def delete_user(db: Session, user_id: int):
|
|
user = user_repository.get(db, user_id)
|
|
if not user:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Pengguna tidak ditemukan")
|
|
if user.username == "admin" or user.id == 1:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Akun Super Admin utama tidak boleh dihapus")
|
|
return user_repository.delete(db, user_id)
|
|
|