Coverage for opt/mealie/lib/python3.12/site-packages/mealie/routes/admin/admin_management_users.py: 80%

1from functools import cached_property 

2 

3from fastapi import APIRouter, Depends, HTTPException, status 

4from pydantic import UUID4 

5 

6from mealie.core import security 

7from mealie.routes._base import BaseAdminController, controller 

8from mealie.routes._base.mixins import HttpRepo 

9from mealie.schema.response.pagination import PaginationQuery 

10from mealie.schema.response.responses import ErrorResponse 

11from mealie.schema.user.auth import UnlockResults 

12from mealie.schema.user.user import UserIn, UserOut, UserPagination 

13from mealie.schema.user.user_passwords import ForgotPassword, PasswordResetToken 

14from mealie.services.user_services.password_reset_service import PasswordResetService 

15from mealie.services.user_services.user_service import UserService 

16 

17router = APIRouter(prefix="/users") 

18 

19 

20@controller(router) 

21class AdminUserManagementRoutes(BaseAdminController): 

22 @cached_property 

23 def repo(self): 

24 return self.repos.users 

25 

26 # ======================================================================= 

27 # CRUD Operations 

28 

29 @property 

30 def mixins(self): 

31 return HttpRepo[UserIn, UserOut, UserOut](self.repo, self.logger, self.registered_exceptions) 

32 

33 @router.get("", response_model=UserPagination) 

34 def get_all(self, q: PaginationQuery = Depends(PaginationQuery)): 

35 response = self.repo.page_all( 

36 pagination=q, 

37 override=UserOut, 

38 ) 

39 

40 response.set_pagination_guides(router.url_path_for("get_all"), q.model_dump()) 

41 return response 

42 

43 @router.post("", response_model=UserOut, status_code=201) 

44 def create_one(self, data: UserIn): 

45 if self.repos.users.get_by_username(data.username): 

46 raise HTTPException(status.HTTP_409_CONFLICT, {"message": self.t("exceptions.username-conflict-error")}) 

47 elif self.repos.users.get_one(data.email, "email"): 

48 raise HTTPException(status.HTTP_409_CONFLICT, {"message": self.t("exceptions.email-conflict-error")}) 

49 

50 data.password = security.hash_password(data.password) 

51 return self.mixins.create_one(data) 

52 

53 @router.post("/unlock", response_model=UnlockResults) 

54 def unlock_users(self, force: bool = False) -> UnlockResults: 

55 user_service = UserService(self.repos) 

56 unlocked = user_service.reset_locked_users(force=force) 

57 

58 return UnlockResults(unlocked=unlocked) 

59 

60 @router.get("/{item_id}", response_model=UserOut) 

61 def get_one(self, item_id: UUID4): 

62 return self.mixins.get_one(item_id) 

63 

64 @router.put("/{item_id}", response_model=UserOut) 

65 def update_one(self, item_id: UUID4, data: UserOut): 

66 # Prevent self demotion 

67 if self.user.id == item_id and self.user.admin != data.admin: 

68 raise HTTPException(status_code=403, detail=ErrorResponse.respond("you cannot demote yourself")) 

69 

70 return self.mixins.update_one(data, item_id) 

71 

72 @router.delete("/{item_id}", response_model=UserOut) 

73 def delete_one(self, item_id: UUID4): 

74 return self.mixins.delete_one(item_id) 

75 

76 @router.post("/password-reset-token", response_model=PasswordResetToken, status_code=201) 

77 def generate_token(self, email: ForgotPassword): 

78 """Generates a reset token and returns it. This is an authenticated endpoint""" 

79 f_service = PasswordResetService(self.session) 

80 token_entry = f_service.generate_reset_token(email.email) 

81 if not token_entry: 

82 raise HTTPException(status_code=500, detail=ErrorResponse.respond("error while generating reset token")) 

83 return PasswordResetToken(token=token_entry.token)