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

1import re 1 ctx1b

2from collections.abc import Iterable 1 ctx1b

3 

4from fastapi import HTTPException, status 1 ctx1b

5from pydantic import UUID4 1 ctx1b

6from slugify import slugify 1 ctx1b

7from sqlalchemy.exc import IntegrityError 1 ctx1b

8 

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

10from mealie.repos.repository_generic import HouseholdRepositoryGeneric 1 ctx1b

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

12from mealie.schema.response.responses import ErrorResponse 1 ctx1b

13 

14 

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

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

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

18 data = SaveCookBook(**data) 

19 data.slug = slugify(data.name) 13 ctx1ghcdiejklmanf

20 

21 max_retries = 10 13 ctx1ghcdiejklmanf

22 for i in range(max_retries): 22 ↛ 29line 22 didn't jump to line 29 because the loop on line 22 didn't complete13 ctx1ghcdiejklmanf

23 try: 13 ctx1ghcdiejklmanf

24 return super().create(data) 13 ctx1ghcdiejklmanf

25 except IntegrityError: 5 ctx1cdeaf

26 self.session.rollback() 5 ctx1cdeaf

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

28 

29 raise # raise the last IntegrityError 

30 

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

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

33 

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

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

36 data = SaveCookBook(**data) 

37 

38 new_slug = slugify(data.name) 1 ctx1a

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 true1 ctx1a

40 data.slug = new_slug 1 ctx1a

41 

42 max_retries = 10 1 ctx1a

43 for i in range(max_retries): 43 ↛ 50line 43 didn't jump to line 50 because the loop on line 43 didn't complete1 ctx1a

44 try: 1 ctx1a

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

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

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: 1 ctx1b

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)