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

50 statements  

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

1from collections.abc import Iterable 1a

2from typing import cast 1a

3from uuid import UUID 1a

4 

5from pydantic import UUID4 1a

6from slugify import slugify 1a

7from sqlalchemy import select 1a

8from sqlalchemy.exc import IntegrityError 1a

9 

10from mealie.db.models.group import Group 1a

11from mealie.schema.user.user import GroupBase, GroupInDB, UpdateGroup 1a

12 

13from .repository_generic import RepositoryGeneric 1a

14 

15 

16class RepositoryGroup(RepositoryGeneric[GroupInDB, Group]): 1a

17 def create(self, data: GroupBase | dict) -> GroupInDB: 1a

18 if isinstance(data, GroupBase): 18 ↛ 21line 18 didn't jump to line 21 because the condition on line 18 was always true1ab

19 data = data.model_dump() 1ab

20 

21 max_attempts = 10 1ab

22 original_name = cast(str, data["name"]) 1ab

23 

24 attempts = 0 1ab

25 while True: 1ab

26 try: 1ab

27 data["slug"] = slugify(data["name"]) 1ab

28 return super().create(data) 1ab

29 except IntegrityError: 1b

30 self.session.rollback() 1b

31 attempts += 1 1b

32 if attempts >= max_attempts: 32 ↛ 33line 32 didn't jump to line 33 because the condition on line 32 was never true1b

33 raise 

34 

35 data["name"] = f"{original_name} ({attempts})" 1b

36 

37 def create_many(self, data: Iterable[GroupInDB | dict]) -> list[GroupInDB]: 1a

38 # since create uses special logic for resolving slugs, we don't want to use the standard create_many method 

39 return [self.create(new_group) for new_group in data] 

40 

41 def update(self, match_value: str | int | UUID4, new_data: UpdateGroup | dict) -> GroupInDB: 1a

42 if isinstance(new_data, GroupBase): 

43 new_data.slug = slugify(new_data.name) 

44 else: 

45 new_data["slug"] = slugify(new_data["name"]) 

46 

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

48 

49 def update_many(self, data: Iterable[UpdateGroup | dict]) -> list[GroupInDB]: 1a

50 # since update uses special logic for resolving slugs, we don't want to use the standard update_many method 

51 return [self.update(group["id"] if isinstance(group, dict) else group.id, group) for group in data] 

52 

53 def get_by_name(self, name: str) -> GroupInDB | None: 1a

54 dbgroup = self.session.execute(select(self.model).filter_by(name=name)).scalars().one_or_none() 1cdefghij

55 if dbgroup is None: 55 ↛ 56line 55 didn't jump to line 56 because the condition on line 55 was never true1cdefghij

56 return None 

57 return self.schema.model_validate(dbgroup) 1cdefghij

58 

59 def get_by_slug_or_id(self, slug_or_id: str | UUID) -> GroupInDB | None: 1a

60 if isinstance(slug_or_id, str): 60 ↛ 66line 60 didn't jump to line 66 because the condition on line 60 was always true1c

61 try: 1c

62 slug_or_id = UUID(slug_or_id) 1c

63 except ValueError: 1c

64 pass 1c

65 

66 if isinstance(slug_or_id, UUID): 66 ↛ 67line 66 didn't jump to line 67 because the condition on line 66 was never true1c

67 return self.get_one(slug_or_id) 

68 else: 

69 return self.get_one(slug_or_id, key="slug") 1c