Coverage for opt/mealie/lib/python3.12/site-packages/mealie/schema/_mealie/mealie_model.py: 74%
90 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-11-25 15:48 +0000
1from __future__ import annotations 1a
3import re 1a
4from collections.abc import Sequence 1a
5from datetime import UTC, datetime 1a
6from enum import Enum 1a
7from typing import ClassVar, Protocol, Self 1a
9from humps.main import camelize 1a
10from pydantic import UUID4, AliasChoices, BaseModel, ConfigDict, Field, model_validator 1a
11from sqlalchemy import Select, desc, func, or_, text 1a
12from sqlalchemy.orm import InstrumentedAttribute, Session 1a
13from sqlalchemy.orm.interfaces import LoaderOption 1a
15from mealie.db.models._model_base import SqlAlchemyBase 1a
17HOUR_ONLY_TZ_PATTERN = re.compile(r"[+-]\d{2}$") 1a
20def UpdatedAtField(*args, **kwargs): 1a
21 """
22 Wrapper for Pydantic's Field, which sets default values for our update_at aliases
24 Since the database stores this value as update_at, we want to accept this as a possible value
25 """
27 kwargs.pop("alias", None) 1a
28 kwargs.pop("alias_generator", None) 1a
30 # ensure the alias is not overwritten by the generator, if there is one
31 # https://docs.pydantic.dev/latest/concepts/alias/#alias-priority
32 kwargs["alias_priority"] = 2 1a
34 kwargs["validation_alias"] = AliasChoices("update_at", "updateAt", "updated_at", "updatedAt") 1a
35 kwargs["serialization_alias"] = "updatedAt" 1a
37 return Field(*args, **kwargs) 1a
40class SearchType(Enum): 1a
41 fuzzy = "fuzzy" 1a
42 tokenized = "tokenized" 1a
45class MealieModel(BaseModel): 1a
46 _fuzzy_similarity_threshold: ClassVar[float] = 0.5 1a
47 _normalize_search: ClassVar[bool] = False 1a
48 _searchable_properties: ClassVar[list[str]] = [] 1a
49 """ 1a
50 Searchable properties for the search API.
51 The first property will be used for sorting (order_by)
52 """
53 model_config = ConfigDict(alias_generator=camelize, populate_by_name=True) 1a
55 @model_validator(mode="before") 1a
56 @classmethod 1a
57 def fix_hour_only_tz[T: BaseModel](cls, data: T) -> T: 1a
58 """
59 Fixes datetimes with timezones that only have the hour portion.
61 Pydantic assumes timezones are in the format +HH:MM, but postgres returns +HH.
62 https://github.com/pydantic/pydantic/issues/8609
63 """
64 for field, field_info in cls.model_fields.items(): 1auxpqyzvrsABCDwcdefghmijknoltb
65 if field_info.annotation != datetime: 1auxpqyzvrsABCDwcdefghmijknoltb
66 continue 1auxpqyzvrsABCDwcdefghmijknoltb
67 try: 1rscdefghmijknolb
68 if not isinstance(val := getattr(data, field), str): 1rscdefghmijknolb
69 continue 1rcdefghijklb
70 except AttributeError: 1rscdefghmijknolb
71 continue 1rscdefghmijknolb
72 if re.search(HOUR_ONLY_TZ_PATTERN, val): 72 ↛ 73line 72 didn't jump to line 73 because the condition on line 72 was never true
73 setattr(data, field, val + ":00")
75 return data 1auxpqyzvrsABCDwcdefghmijknoltb
77 @model_validator(mode="after") 1a
78 def set_tz_info(self) -> Self: 1a
79 """
80 Adds UTC timezone information to all datetimes in the model.
81 The server stores everything in UTC without timezone info.
82 """
83 for field in self.__class__.model_fields: 1auxpqyzvrsABCDwcdefghmijknoltb
84 val = getattr(self, field) 1auxpqyzvrsABCDwcdefghmijknoltb
85 if not isinstance(val, datetime): 1auxpqyzvrsABCDwcdefghmijknoltb
86 continue 1auxpqyzvrsABCDwcdefghmijknoltb
87 if not val.tzinfo: 1yzrsABCDwcdefghmijknoltb
88 setattr(self, field, val.replace(tzinfo=UTC))
90 return self 1auxpqyzvrsABCDwcdefghmijknoltb
92 def cast[T: BaseModel](self, cls: type[T], **kwargs) -> T: 1a
93 """
94 Cast the current model to another with additional arguments. Useful for
95 transforming DTOs into models that are saved to a database
96 """
97 create_data = { 1auvcdefghmijknoltb
98 field: getattr(self, field) for field in self.__class__.model_fields if field in cls.model_fields
99 }
100 create_data.update(kwargs or {}) 1auvcdefghmijknoltb
101 return cls(**create_data) 1auvcdefghmijknoltb
103 def map_to[T: BaseModel](self, dest: T) -> T: 1a
104 """
105 Map matching values from the current model to another model. Model returned
106 for method chaining.
107 """
109 for field in self.__class__.model_fields:
110 if field in dest.__class__.model_fields:
111 setattr(dest, field, getattr(self, field))
113 return dest
115 def map_from(self, src: BaseModel): 1a
116 """
117 Map matching values from another model to the current model.
118 """
120 for field in src.__class__.model_fields:
121 if field in self.__class__.model_fields:
122 setattr(self, field, getattr(src, field))
124 def merge[T: BaseModel](self, src: T, replace_null=False): 1a
125 """
126 Replace matching values from another instance to the current instance.
127 """
129 for field in src.__class__.model_fields:
130 val = getattr(src, field)
131 if field in self.__class__.model_fields and (val is not None or replace_null):
132 setattr(self, field, val)
134 @classmethod 1a
135 def loader_options(cls) -> list[LoaderOption]: 1a
136 return [] 1xpqwcdefghmijknoltb
138 @classmethod 1a
139 def filter_search_query( 1a
140 cls,
141 db_model: type[SqlAlchemyBase],
142 query: Select,
143 session: Session,
144 search_type: SearchType,
145 search: str,
146 search_list: list[str],
147 ) -> Select:
148 """
149 Filters a search query based on model attributes
151 Can be overridden to support a more advanced search
152 """
154 if not cls._searchable_properties: 154 ↛ 155line 154 didn't jump to line 155 because the condition on line 154 was never true1pqcdefghmijknolb
155 raise AttributeError("Not Implemented")
157 model_properties: list[InstrumentedAttribute] = [getattr(db_model, prop) for prop in cls._searchable_properties] 1pqcdefghmijknolb
158 if search_type is SearchType.fuzzy: 158 ↛ 159line 158 didn't jump to line 159 because the condition on line 158 was never true1pqcdefghmijknolb
159 session.execute(text(f"set pg_trgm.word_similarity_threshold = {cls._fuzzy_similarity_threshold};"))
160 filters = [prop.op("%>")(search) for prop in model_properties]
162 # trigram ordering by the first searchable property
163 return query.filter(or_(*filters)).order_by(func.least(model_properties[0].op("<->>")(search)))
164 else:
165 filters = [] 1pqcdefghmijknolb
166 for prop in model_properties: 1pqcdefghmijknolb
167 filters.extend([prop.like(f"%{s}%") for s in search_list]) 1pqcdefghmijknolb
169 # order by how close the result is to the first searchable property
170 return query.filter(or_(*filters)).order_by(desc(model_properties[0].like(f"%{search}%"))) 1pqcdefghmijknolb
173class HasUUID(Protocol): 1a
174 id: UUID4 1a
177def extract_uuids(models: Sequence[HasUUID]) -> list[UUID4]: 1a
178 return [x.id for x in models]