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

1from collections.abc import Iterable 

2from typing import cast 

3from uuid import UUID 

4 

5from pydantic import UUID4 

6from slugify import slugify 

7from sqlalchemy import select 

8from sqlalchemy.exc import IntegrityError 

9 

10from mealie.db.models.group import Group 

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

12 

13from .repository_generic import RepositoryGeneric 

14 

15 

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

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

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

19 data = data.model_dump() 

20 

21 max_attempts = 10 

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

23 

24 attempts = 0 

25 while True: 

26 try: 

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

28 return super().create(data) 

29 except IntegrityError: 

30 self.session.rollback() 

31 attempts += 1 

32 if attempts >= max_attempts: 

33 raise 

34 

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

36 

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

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: 

42 if isinstance(new_data, GroupBase): 42 ↛ 45line 42 didn't jump to line 45 because the condition on line 42 was always true

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]: 

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: 

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

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

56 return None 

57 return self.schema.model_validate(dbgroup) 

58 

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

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

61 try: 

62 slug_or_id = UUID(slug_or_id) 

63 except ValueError: 

64 pass 

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 true

67 return self.get_one(slug_or_id) 

68 else: 

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