Coverage for opt/mealie/lib/python3.12/site-packages/mealie/repos/repository_group.py: 74%
50 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-11-25 15:48 +0000
« prev ^ index » next coverage.py v7.10.6, created at 2025-11-25 15:48 +0000
1from collections.abc import Iterable 1a
2from typing import cast 1a
3from uuid import UUID 1a
5from pydantic import UUID4 1a
6from slugify import slugify 1a
7from sqlalchemy import select 1a
8from sqlalchemy.exc import IntegrityError 1a
10from mealie.db.models.group import Group 1a
11from mealie.schema.user.user import GroupBase, GroupInDB, UpdateGroup 1a
13from .repository_generic import RepositoryGeneric 1a
16class RepositoryGroup(RepositoryGeneric[GroupInDB, Group]): 1a
17 def create(self, data: GroupBase | dict) -> GroupInDB: 1a
18 if isinstance(data, GroupBase): 18 ↛ 21line 18 didn't jump to line 21 because the condition on line 18 was always true1ab
19 data = data.model_dump() 1ab
21 max_attempts = 10 1ab
22 original_name = cast(str, data["name"]) 1ab
24 attempts = 0 1ab
25 while True: 1ab
26 try: 1ab
27 data["slug"] = slugify(data["name"]) 1ab
28 return super().create(data) 1ab
29 except IntegrityError: 1b
30 self.session.rollback() 1b
31 attempts += 1 1b
32 if attempts >= max_attempts: 32 ↛ 33line 32 didn't jump to line 33 because the condition on line 32 was never true1b
33 raise
35 data["name"] = f"{original_name} ({attempts})" 1b
37 def create_many(self, data: Iterable[GroupInDB | dict]) -> list[GroupInDB]: 1a
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]
41 def update(self, match_value: str | int | UUID4, new_data: UpdateGroup | dict) -> GroupInDB: 1a
42 if isinstance(new_data, GroupBase):
43 new_data.slug = slugify(new_data.name)
44 else:
45 new_data["slug"] = slugify(new_data["name"])
47 return super().update(match_value, new_data)
49 def update_many(self, data: Iterable[UpdateGroup | dict]) -> list[GroupInDB]: 1a
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]
53 def get_by_name(self, name: str) -> GroupInDB | None: 1a
54 dbgroup = self.session.execute(select(self.model).filter_by(name=name)).scalars().one_or_none() 1efghijklmnopqrstuvwc
55 if dbgroup is None: 55 ↛ 56line 55 didn't jump to line 56 because the condition on line 55 was never true1efghijklmnopqrstuvwc
56 return None
57 return self.schema.model_validate(dbgroup) 1efghijklmnopqrstuvwc
59 def get_by_slug_or_id(self, slug_or_id: str | UUID) -> GroupInDB | None: 1a
60 if isinstance(slug_or_id, str): 60 ↛ 66line 60 didn't jump to line 66 because the condition on line 60 was always true1dc
61 try: 1dc
62 slug_or_id = UUID(slug_or_id) 1dc
63 except ValueError: 1dc
64 pass 1dc
66 if isinstance(slug_or_id, UUID): 66 ↛ 67line 66 didn't jump to line 67 because the condition on line 66 was never true1dc
67 return self.get_one(slug_or_id)
68 else:
69 return self.get_one(slug_or_id, key="slug") 1dc