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

49 statements  

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

1import re 1b

2from collections.abc import Iterable 1b

3 

4from fastapi import HTTPException, status 1b

5from pydantic import UUID4 1b

6from slugify import slugify 1b

7from sqlalchemy.exc import IntegrityError 1b

8 

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

10from mealie.repos.repository_generic import HouseholdRepositoryGeneric 1b

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

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

13 

14 

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

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

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

18 data = SaveCookBook(**data) 

19 data.slug = slugify(data.name) 1lmcrdenfostuavpgwxhiqjyzAk

20 

21 max_retries = 10 1lmcrdenfostuavpgwxhiqjyzAk

22 for i in range(max_retries): 1lmcrdenfostuavpgwxhiqjyzAk

23 try: 1lmcrdenfostuavpgwxhiqjyzAk

24 return super().create(data) 1lmcrdenfostuavpgwxhiqjyzAk

25 except IntegrityError: 1lmcdenfoapghiqjk

26 self.session.rollback() 1lmcdenfoapghiqjk

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

28 

29 raise # raise the last IntegrityError 1cdefaghijk

30 

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

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

33 

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

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

36 data = SaveCookBook(**data) 

37 

38 new_slug = slugify(data.name) 1a

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 true1a

40 data.slug = new_slug 1a

41 

42 max_retries = 10 1a

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

44 try: 1a

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

46 except IntegrityError: 1a

47 self.session.rollback() 1a

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

49 

50 raise # raise the last IntegrityError 

51 

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

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

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)