Coverage for opt/mealie/lib/python3.12/site-packages/mealie/routes/households/controller_invitations.py: 89%

34 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-11-25 15:48 +0000

1from typing import Annotated 1a

2 

3from fastapi import APIRouter, Header, HTTPException, status 1a

4 

5from mealie.core.security import url_safe_token 1a

6from mealie.routes._base import BaseUserController, controller 1a

7from mealie.schema.household.invite_token import ( 1a

8 CreateInviteToken, 

9 EmailInitationResponse, 

10 EmailInvitation, 

11 ReadInviteToken, 

12 SaveInviteToken, 

13) 

14from mealie.schema.response.pagination import PaginationQuery 1a

15from mealie.services.email.email_service import EmailService 1a

16 

17router = APIRouter(prefix="/households/invitations", tags=["Households: Invitations"]) 1a

18 

19 

20@controller(router) 1a

21class GroupInvitationsController(BaseUserController): 1a

22 @router.get("", response_model=list[ReadInviteToken]) 1a

23 def get_invite_tokens(self): 1a

24 return self.repos.group_invite_tokens.page_all(PaginationQuery(page=1, per_page=-1)).items 

25 

26 @router.post("", response_model=ReadInviteToken, status_code=status.HTTP_201_CREATED) 1a

27 def create_invite_token(self, body: CreateInviteToken): 1a

28 if not self.user.can_invite: 28 ↛ 29line 28 didn't jump to line 29 because the condition on line 28 was never true

29 raise HTTPException( 

30 status.HTTP_403_FORBIDDEN, 

31 detail="User is not allowed to create invite tokens", 

32 ) 

33 

34 body.group_id = body.group_id or self.group_id 

35 body.household_id = body.household_id or self.household_id 

36 

37 if not self.user.admin and (body.group_id != self.group_id or body.household_id != self.household_id): 

38 raise HTTPException( 

39 status.HTTP_403_FORBIDDEN, 

40 detail="Only admins can create invite tokens for other groups or households", 

41 ) 

42 

43 token = SaveInviteToken( 

44 uses_left=body.uses, group_id=body.group_id, household_id=body.household_id, token=url_safe_token() 

45 ) 

46 return self.repos.group_invite_tokens.create(token) 

47 

48 @router.post("/email", response_model=EmailInitationResponse) 1a

49 def email_invitation( 1a

50 self, 

51 invite: EmailInvitation, 

52 accept_language: Annotated[str | None, Header()] = None, 

53 ): 

54 email_service = EmailService(locale=accept_language) 

55 url = f"{self.settings.BASE_URL}/register?token={invite.token}" 

56 

57 success = False 

58 error = None 

59 try: 

60 success = email_service.send_invitation(address=invite.email, invitation_url=url) 

61 except Exception as e: 

62 error = str(e) 

63 

64 return EmailInitationResponse(success=success, error=error)