mirror of
https://github.com/mealie-recipes/mealie.git
synced 2025-07-18 20:59:41 +02:00
306 lines
8.1 KiB
Python
306 lines
8.1 KiB
Python
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import Annotated, Any, Generic, TypeVar
|
|
from uuid import UUID
|
|
|
|
from pydantic import UUID4, BaseModel, ConfigDict, Field, StringConstraints, field_validator
|
|
from sqlalchemy.orm import joinedload, selectinload
|
|
from sqlalchemy.orm.interfaces import LoaderOption
|
|
|
|
from mealie.core.config import get_app_dirs, get_app_settings
|
|
from mealie.db.models.recipe.recipe import RecipeModel
|
|
from mealie.db.models.users import User
|
|
from mealie.db.models.users.user_to_recipe import UserToRecipe
|
|
from mealie.db.models.users.users import AuthMethod, LongLiveToken
|
|
from mealie.schema._mealie import MealieModel
|
|
from mealie.schema.group.group_preferences import ReadGroupPreferences
|
|
from mealie.schema.household.webhook import CreateWebhook, ReadWebhook
|
|
from mealie.schema.response.pagination import PaginationBase
|
|
|
|
from ...db.models.group import Group
|
|
from ..recipe import CategoryBase
|
|
|
|
DataT = TypeVar("DataT", bound=BaseModel)
|
|
DEFAULT_INTEGRATION_ID = "generic"
|
|
settings = get_app_settings()
|
|
|
|
|
|
class LongLiveTokenIn(MealieModel):
|
|
name: str
|
|
integration_id: str = DEFAULT_INTEGRATION_ID
|
|
|
|
|
|
class LongLiveTokenOut(MealieModel):
|
|
token: str
|
|
name: str
|
|
id: int
|
|
created_at: datetime | None = None
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
@classmethod
|
|
def loader_options(cls) -> list[LoaderOption]:
|
|
return [joinedload(LongLiveToken.user)]
|
|
|
|
|
|
class CreateToken(LongLiveTokenIn):
|
|
user_id: UUID4
|
|
token: str
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
|
|
class DeleteTokenResponse(MealieModel):
|
|
token_delete: str
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
|
|
class ChangePassword(MealieModel):
|
|
current_password: str = ""
|
|
new_password: str = Field(..., min_length=8)
|
|
|
|
|
|
class GroupBase(MealieModel):
|
|
name: Annotated[str, StringConstraints(strip_whitespace=True, min_length=1)]
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
|
|
class UserRatingSummary(MealieModel):
|
|
recipe_id: UUID4
|
|
rating: float | None = None
|
|
is_favorite: Annotated[bool, Field(validate_default=True)] = False
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
@field_validator("is_favorite", mode="before")
|
|
def convert_is_favorite(cls, v: Any) -> bool:
|
|
if v is None:
|
|
return False
|
|
else:
|
|
return v
|
|
|
|
|
|
class UserRatingCreate(UserRatingSummary):
|
|
user_id: UUID4
|
|
|
|
|
|
class UserRatingUpdate(MealieModel):
|
|
rating: float | None = None
|
|
is_favorite: bool | None = None
|
|
|
|
|
|
class UserRatingOut(UserRatingCreate):
|
|
id: UUID4
|
|
|
|
@classmethod
|
|
def loader_options(cls) -> list[LoaderOption]:
|
|
return [
|
|
joinedload(UserToRecipe.recipe).joinedload(RecipeModel.user).load_only(User.household_id, User.group_id)
|
|
]
|
|
|
|
|
|
class UserRatings(BaseModel, Generic[DataT]):
|
|
ratings: list[DataT]
|
|
|
|
|
|
class UserBase(MealieModel):
|
|
id: UUID4 | None = None
|
|
username: str | None = None
|
|
full_name: str | None = None
|
|
email: Annotated[str, StringConstraints(to_lower=True, strip_whitespace=True)]
|
|
auth_method: AuthMethod = AuthMethod.MEALIE
|
|
admin: bool = False
|
|
group: str | None = None
|
|
household: str | None = None
|
|
advanced: bool = False
|
|
|
|
can_invite: bool = False
|
|
can_manage: bool = False
|
|
can_manage_household: bool = False
|
|
can_organize: bool = False
|
|
model_config = ConfigDict(
|
|
from_attributes=True,
|
|
json_schema_extra={
|
|
"example": {
|
|
"username": "ChangeMe",
|
|
"fullName": "Change Me",
|
|
"email": "changeme@example.com",
|
|
"group": settings.DEFAULT_GROUP,
|
|
"household": settings.DEFAULT_HOUSEHOLD,
|
|
"admin": "false",
|
|
}
|
|
},
|
|
)
|
|
|
|
@field_validator("group", mode="before")
|
|
def convert_group_to_name(cls, v):
|
|
if not v or isinstance(v, str):
|
|
return v
|
|
|
|
try:
|
|
return v.name
|
|
except AttributeError:
|
|
return v
|
|
|
|
@field_validator("household", mode="before")
|
|
def convert_household_to_name(cls, v):
|
|
if not v or isinstance(v, str):
|
|
return v
|
|
|
|
try:
|
|
return v.name
|
|
except AttributeError:
|
|
return v
|
|
|
|
|
|
class UserIn(UserBase):
|
|
username: str
|
|
full_name: str
|
|
password: str
|
|
|
|
|
|
class UserOut(UserBase):
|
|
id: UUID4
|
|
group: str
|
|
group_id: UUID4
|
|
group_slug: str
|
|
household: str
|
|
household_id: UUID4
|
|
household_slug: str
|
|
tokens: list[LongLiveTokenOut] | None = None
|
|
cache_key: str
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
@property
|
|
def is_default_user(self) -> bool:
|
|
return self.email == settings._DEFAULT_EMAIL.strip().lower()
|
|
|
|
@classmethod
|
|
def loader_options(cls) -> list[LoaderOption]:
|
|
return [joinedload(User.group), joinedload(User.household), joinedload(User.tokens)]
|
|
|
|
|
|
class UserSummary(MealieModel):
|
|
id: UUID4
|
|
group_id: UUID4
|
|
household_id: UUID4
|
|
username: str
|
|
full_name: str
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
|
|
class UserPagination(PaginationBase):
|
|
items: list[UserOut]
|
|
|
|
|
|
class UserSummaryPagination(PaginationBase):
|
|
items: list[UserSummary]
|
|
|
|
|
|
class PrivateUser(UserOut):
|
|
password: str
|
|
login_attemps: int = 0
|
|
locked_at: datetime | None = None
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
@field_validator("login_attemps", mode="before")
|
|
@classmethod
|
|
def none_to_zero(cls, v):
|
|
return 0 if v is None else v
|
|
|
|
@staticmethod
|
|
def get_directory(user_id: UUID4 | str) -> Path:
|
|
user_dir = get_app_dirs().USER_DIR / str(user_id)
|
|
user_dir.mkdir(parents=True, exist_ok=True)
|
|
return user_dir
|
|
|
|
@property
|
|
def is_locked(self) -> bool:
|
|
if self.locked_at is None:
|
|
return False
|
|
|
|
lockout_expires_at = self.locked_at + timedelta(hours=get_app_settings().SECURITY_USER_LOCKOUT_TIME)
|
|
return lockout_expires_at > datetime.now(UTC)
|
|
|
|
def directory(self) -> Path:
|
|
return PrivateUser.get_directory(self.id)
|
|
|
|
@classmethod
|
|
def loader_options(cls) -> list[LoaderOption]:
|
|
return [joinedload(User.group), joinedload(User.household), joinedload(User.tokens)]
|
|
|
|
|
|
class UpdateGroup(GroupBase):
|
|
id: UUID4
|
|
name: str
|
|
slug: str
|
|
categories: list[CategoryBase] | None = []
|
|
|
|
webhooks: list[CreateWebhook] = []
|
|
|
|
|
|
class GroupHouseholdSummary(MealieModel):
|
|
id: UUID4
|
|
name: str
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
|
|
class GroupInDB(UpdateGroup):
|
|
households: list[GroupHouseholdSummary] | None = None
|
|
users: list[UserSummary] | None = None
|
|
preferences: ReadGroupPreferences | None = None
|
|
webhooks: list[ReadWebhook] = []
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
@staticmethod
|
|
def get_directory(id: UUID4) -> Path:
|
|
group_dir = get_app_dirs().GROUPS_DIR / str(id)
|
|
group_dir.mkdir(parents=True, exist_ok=True)
|
|
return group_dir
|
|
|
|
@staticmethod
|
|
def get_export_directory(id: UUID) -> Path:
|
|
export_dir = GroupInDB.get_directory(id) / "export"
|
|
export_dir.mkdir(parents=True, exist_ok=True)
|
|
return export_dir
|
|
|
|
@property
|
|
def directory(self) -> Path:
|
|
return GroupInDB.get_directory(self.id)
|
|
|
|
@property
|
|
def exports(self) -> Path:
|
|
return GroupInDB.get_export_directory(self.id)
|
|
|
|
@classmethod
|
|
def loader_options(cls) -> list[LoaderOption]:
|
|
return [
|
|
joinedload(Group.categories),
|
|
joinedload(Group.webhooks),
|
|
joinedload(Group.preferences),
|
|
joinedload(Group.households),
|
|
selectinload(Group.users).joinedload(User.group),
|
|
selectinload(Group.users).joinedload(User.tokens),
|
|
]
|
|
|
|
|
|
class GroupSummary(GroupBase):
|
|
id: UUID4
|
|
name: str
|
|
slug: str
|
|
preferences: ReadGroupPreferences | None = None
|
|
|
|
@classmethod
|
|
def loader_options(cls) -> list[LoaderOption]:
|
|
return [
|
|
joinedload(Group.preferences),
|
|
]
|
|
|
|
|
|
class GroupPagination(PaginationBase):
|
|
items: list[GroupInDB]
|
|
|
|
|
|
class LongLiveTokenInDB(CreateToken):
|
|
id: int
|
|
user: PrivateUser
|
|
model_config = ConfigDict(from_attributes=True)
|