Coverage for opt/mealie/lib/python3.12/site-packages/mealie/routes/admin/admin_management_users.py: 52%
58 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-11-25 15:32 +0000
1from functools import cached_property 1a
3from fastapi import APIRouter, Depends, HTTPException, status 1a
4from pydantic import UUID4 1a
6from mealie.core import security 1a
7from mealie.routes._base import BaseAdminController, controller 1a
8from mealie.routes._base.mixins import HttpRepo 1a
9from mealie.schema.response.pagination import PaginationQuery 1a
10from mealie.schema.response.responses import ErrorResponse 1a
11from mealie.schema.user.auth import UnlockResults 1a
12from mealie.schema.user.user import UserIn, UserOut, UserPagination 1a
13from mealie.schema.user.user_passwords import ForgotPassword, PasswordResetToken 1a
14from mealie.services.user_services.password_reset_service import PasswordResetService 1a
15from mealie.services.user_services.user_service import UserService 1a
17router = APIRouter(prefix="/users") 1a
20@controller(router) 1a
21class AdminUserManagementRoutes(BaseAdminController): 1a
22 @cached_property 1a
23 def repo(self): 1a
24 return self.repos.users
26 # =======================================================================
27 # CRUD Operations
29 @property 1a
30 def mixins(self): 1a
31 return HttpRepo[UserIn, UserOut, UserOut](self.repo, self.logger, self.registered_exceptions)
33 @router.get("", response_model=UserPagination) 1a
34 def get_all(self, q: PaginationQuery = Depends(PaginationQuery)): 1a
35 response = self.repo.page_all(
36 pagination=q,
37 override=UserOut,
38 )
40 response.set_pagination_guides(router.url_path_for("get_all"), q.model_dump())
41 return response
43 @router.post("", response_model=UserOut, status_code=201) 1a
44 def create_one(self, data: UserIn): 1a
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")})
50 data.password = security.hash_password(data.password)
51 return self.mixins.create_one(data)
53 @router.post("/unlock", response_model=UnlockResults) 1a
54 def unlock_users(self, force: bool = False) -> UnlockResults: 1a
55 user_service = UserService(self.repos)
56 unlocked = user_service.reset_locked_users(force=force)
58 return UnlockResults(unlocked=unlocked)
60 @router.get("/{item_id}", response_model=UserOut) 1a
61 def get_one(self, item_id: UUID4): 1a
62 return self.mixins.get_one(item_id)
64 @router.put("/{item_id}", response_model=UserOut) 1a
65 def update_one(self, item_id: UUID4, data: UserOut): 1a
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"))
70 return self.mixins.update_one(data, item_id)
72 @router.delete("/{item_id}", response_model=UserOut) 1a
73 def delete_one(self, item_id: UUID4): 1a
74 return self.mixins.delete_one(item_id)
76 @router.post("/password-reset-token", response_model=PasswordResetToken, status_code=201) 1a
77 def generate_token(self, email: ForgotPassword): 1a
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)