Coverage for opt/mealie/lib/python3.12/site-packages/mealie/repos/repository_cookbooks.py: 43%
49 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 13:45 +0000
1import re 1a
2from collections.abc import Iterable 1a
4from fastapi import HTTPException, status 1a
5from pydantic import UUID4 1a
6from slugify import slugify 1a
7from sqlalchemy.exc import IntegrityError 1a
9from mealie.db.models.household.cookbook import CookBook 1a
10from mealie.repos.repository_generic import HouseholdRepositoryGeneric 1a
11from mealie.schema.cookbook.cookbook import ReadCookBook, SaveCookBook 1a
12from mealie.schema.response.responses import ErrorResponse 1a
15class RepositoryCookbooks(HouseholdRepositoryGeneric[ReadCookBook, CookBook]): 1a
16 def create(self, data: SaveCookBook | dict) -> ReadCookBook: 1a
17 if isinstance(data, dict): 17 ↛ 18line 17 didn't jump to line 18 because the condition on line 17 was never true1ghibcjdeklmf
18 data = SaveCookBook(**data)
19 data.slug = slugify(data.name) 1ghibcjdeklmf
21 max_retries = 10 1ghibcjdeklmf
22 for i in range(max_retries): 22 ↛ 29line 22 didn't jump to line 29 because the loop on line 22 didn't complete1ghibcjdeklmf
23 try: 1ghibcjdeklmf
24 return super().create(data) 1ghibcjdeklmf
25 except IntegrityError: 1bcdef
26 self.session.rollback() 1bcdef
27 data.slug = slugify(f"{data.name} ({i + 1})") 1bcdef
29 raise # raise the last IntegrityError
31 def create_many(self, data: Iterable[ReadCookBook | dict]) -> list[ReadCookBook]: 1a
32 return [self.create(entry) for entry in data]
34 def update(self, match_value: str | int | UUID4, data: SaveCookBook | dict) -> ReadCookBook: 1a
35 if isinstance(data, dict):
36 data = SaveCookBook(**data)
38 new_slug = slugify(data.name)
39 if not (data.slug and re.match(rf"^({new_slug})(-\d+)?$", data.slug)):
40 data.slug = new_slug
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})")
50 raise # raise the last IntegrityError
52 def update_many(self, data: Iterable[ReadCookBook | dict]) -> list[ReadCookBook]: 1a
53 return [self.update(entry.id if isinstance(entry, ReadCookBook) else entry["id"], entry) for entry in data]
55 def patch(self, match_value: str | int | UUID4, data: SaveCookBook | dict) -> ReadCookBook: 1a
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()
64 if not isinstance(data, dict):
65 data = data.model_dump()
66 return self.update(match_value, cookbook_data | data)