Coverage for opt/mealie/lib/python3.12/site-packages/mealie/repos/repository_users.py: 56%

1import random 

2import shutil 

3 

4from pydantic import UUID4 

5from sqlalchemy import select 

6 

7from mealie.assets import users as users_assets 

8from mealie.core.config import get_app_settings 

9from mealie.db.models.users.user_to_recipe import UserToRecipe 

10from mealie.schema.user.user import PrivateUser, UserRatingOut 

11 

12from ..db.models.users import User 

13from .repository_generic import GroupRepositoryGeneric 

14 

15settings = get_app_settings() 

16 

17 

18class RepositoryUsers(GroupRepositoryGeneric[PrivateUser, User]): 

19 def update_password(self, id, password: str): 

20 entry = self._query_one(match_value=id) 

21 if settings.IS_DEMO: 

22 user_to_update = self.schema.model_validate(entry) 

23 if user_to_update.is_default_user: 

24 # do not update the default user in demo mode 

25 return user_to_update 

26 

27 entry.update_password(password) 

28 self.session.commit() 

29 

30 return self.schema.model_validate(entry) 

31 

32 def create(self, user: PrivateUser | dict): # type: ignore 

33 new_user = super().create(user) 

34 

35 # Select Random Image 

36 all_images = [ 

37 users_assets.img_random_1, 

38 users_assets.img_random_2, 

39 users_assets.img_random_3, 

40 ] 

41 random_image = random.choice(all_images) 

42 shutil.copy(random_image, new_user.directory() / "profile.webp") 

43 

44 return new_user 

45 

46 def update(self, match_value: str | int | UUID4, new_data: dict | PrivateUser) -> PrivateUser: 

47 if settings.IS_DEMO: 47 ↛ 48line 47 didn't jump to line 48 because the condition on line 47 was never true

48 user_to_update = self.get_one(match_value) 

49 if user_to_update and user_to_update.is_default_user: 

50 # do not update the default user in demo mode 

51 return user_to_update 

52 

53 return super().update(match_value, new_data) 

54 

55 def delete(self, value: str | UUID4, match_key: str | None = None) -> User: 

56 if settings.IS_DEMO: 

57 user_to_delete = self.get_one(value, match_key) 

58 if user_to_delete and user_to_delete.is_default_user: 

59 # do not update the default user in demo mode 

60 return user_to_delete 

61 

62 entry = super().delete(value, match_key) 

63 # Delete the user's directory 

64 shutil.rmtree(PrivateUser.get_directory(value)) 

65 return entry 

66 

67 def get_by_username(self, username: str) -> PrivateUser | None: 

68 stmt = select(User).filter(User.username == username) 

69 dbuser = self.session.execute(stmt).scalars().one_or_none() 

70 return None if dbuser is None else self.schema.model_validate(dbuser) 

71 

72 def get_locked_users(self) -> list[PrivateUser]: 

73 stmt = select(User).filter(User.locked_at != None) # noqa E711 

74 results = self.session.execute(stmt).scalars().all() 

75 return [self.schema.model_validate(x) for x in results] 

76 

77 

78class RepositoryUserRatings(GroupRepositoryGeneric[UserRatingOut, UserToRecipe]): 

79 # Since users can post events on recipes that belong to other households, 

80 # this is a group repository, rather than a household repository. 

81 

82 def get_by_user(self, user_id: UUID4, favorites_only=False) -> list[UserRatingOut]: 

83 stmt = select(UserToRecipe).filter(UserToRecipe.user_id == user_id) 

84 if favorites_only: 

85 stmt = stmt.filter(UserToRecipe.is_favorite) 

86 

87 results = self.session.execute(stmt).scalars().all() 

88 return [self.schema.model_validate(x) for x in results] 

89 

90 def get_by_recipe(self, recipe_id: UUID4, favorites_only=False) -> list[UserRatingOut]: 

91 stmt = select(UserToRecipe).filter(UserToRecipe.recipe_id == recipe_id) 

92 if favorites_only: 

93 stmt = stmt.filter(UserToRecipe.is_favorite) 

94 

95 results = self.session.execute(stmt).scalars().all() 

96 return [self.schema.model_validate(x) for x in results] 

97 

98 def get_by_user_and_recipe(self, user_id: UUID4, recipe_id: UUID4) -> UserRatingOut | None: 

99 stmt = select(UserToRecipe).filter(UserToRecipe.user_id == user_id, UserToRecipe.recipe_id == recipe_id) 

100 result = self.session.execute(stmt).scalars().one_or_none() 

101 return None if result is None else self.schema.model_validate(result)