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

1import re 

2from collections.abc import Iterable 

3 

4from fastapi import HTTPException, status 

5from pydantic import UUID4 

6from slugify import slugify 

7from sqlalchemy.exc import IntegrityError 

8 

9from mealie.db.models.household.cookbook import CookBook 

10from mealie.repos.repository_generic import HouseholdRepositoryGeneric 

11from mealie.schema.cookbook.cookbook import ReadCookBook, SaveCookBook 

12from mealie.schema.response.responses import ErrorResponse 

13 

14 

15class RepositoryCookbooks(HouseholdRepositoryGeneric[ReadCookBook, CookBook]): 

16 def create(self, data: SaveCookBook | dict) -> ReadCookBook: 

17 if isinstance(data, dict): 17 ↛ 18line 17 didn't jump to line 18 because the condition on line 17 was never true

18 data = SaveCookBook(**data) 

19 data.slug = slugify(data.name) 

20 

21 max_retries = 10 

22 for i in range(max_retries): 

23 try: 

24 return super().create(data) 

25 except IntegrityError: 

26 self.session.rollback() 

27 data.slug = slugify(f"{data.name} ({i + 1})") 

28 

29 raise # raise the last IntegrityError 

30 

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

32 return [self.create(entry) for entry in data] 

33 

34 def update(self, match_value: str | int | UUID4, data: SaveCookBook | dict) -> ReadCookBook: 

35 if isinstance(data, dict): 35 ↛ 36line 35 didn't jump to line 36 because the condition on line 35 was never true

36 data = SaveCookBook(**data) 

37 

38 new_slug = slugify(data.name) 

39 if not (data.slug and re.match(rf"^({new_slug})(-\d+)?$", data.slug)): 39 ↛ 42line 39 didn't jump to line 42 because the condition on line 39 was always true

40 data.slug = new_slug 

41 

42 max_retries = 10 

43 for i in range(max_retries): 

44 try: 

45 return super().update(match_value, data) 

46 except IntegrityError: 

47 self.session.rollback() 

48 data.slug = slugify(f"{data.name} ({i + 1})") 

49 

50 raise # raise the last IntegrityError 

51 

52 def update_many(self, data: Iterable[ReadCookBook | dict]) -> list[ReadCookBook]: 

53 return [self.update(entry.id if isinstance(entry, ReadCookBook) else entry["id"], entry) for entry in data] 

54 

55 def patch(self, match_value: str | int | UUID4, data: SaveCookBook | dict) -> ReadCookBook: 

56 cookbook = self.get_one(match_value) 

57 if not cookbook: 

58 raise HTTPException( 

59 status.HTTP_404_NOT_FOUND, 

60 detail=ErrorResponse.respond(message="Not found."), 

61 ) 

62 cookbook_data = cookbook.model_dump() 

63 

64 if not isinstance(data, dict): 

65 data = data.model_dump() 

66 return self.update(match_value, cookbook_data | data)