fooder-api/fooder/domain/meal.py

154 lines
4.1 KiB
Python
Raw Normal View History

2023-04-01 16:19:12 +02:00
from sqlalchemy.orm import relationship, Mapped, mapped_column, joinedload
from sqlalchemy import ForeignKey, Integer
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.exc import IntegrityError
from typing import Optional
from .base import Base, CommonMixin
from .entry import Entry
2023-10-27 15:12:48 +02:00
from .preset import Preset
2023-04-01 16:19:12 +02:00
class Meal(Base, CommonMixin):
"""Meal."""
name: Mapped[str]
order: Mapped[int]
diary_id: Mapped[int] = mapped_column(Integer, ForeignKey("diary.id"))
2023-04-03 13:46:03 +02:00
entries: Mapped[list[Entry]] = relationship(
lazy="selectin", order_by=Entry.last_changed
)
2023-04-01 16:19:12 +02:00
@property
def calories(self) -> float:
"""calories.
:rtype: float
"""
return sum(entry.calories for entry in self.entries)
@property
def protein(self) -> float:
"""protein.
:rtype: float
"""
return sum(entry.protein for entry in self.entries)
@property
def carb(self) -> float:
"""carb.
:rtype: float
"""
return sum(entry.carb for entry in self.entries)
@property
def fat(self) -> float:
"""fat.
:rtype: float
"""
return sum(entry.fat for entry in self.entries)
2023-07-30 20:18:42 +02:00
@property
def fiber(self) -> float:
"""fiber.
:rtype: float
"""
return sum(entry.fiber for entry in self.entries)
2023-04-01 16:19:12 +02:00
@classmethod
async def create(
cls,
session: AsyncSession,
diary_id: int,
name: Optional[str] = None,
) -> "Meal":
# check if order already exists in diary
query = (
select(cls.order).where(cls.diary_id == diary_id).order_by(cls.order.desc())
)
2023-04-01 16:19:12 +02:00
existing_meal = await session.scalar(query)
order = existing_meal + 1 if existing_meal else 1
2023-04-01 16:19:12 +02:00
if name is None:
name = f"Meal {order}"
meal = Meal(diary_id=diary_id, name=name, order=order)
session.add(meal)
try:
await session.flush()
except IntegrityError:
raise AssertionError("diary does not exist")
2023-04-03 13:46:03 +02:00
meal = await cls._get_by_id(session, meal.id)
2023-04-01 16:19:12 +02:00
if not meal:
raise RuntimeError()
return meal
2023-10-27 15:12:48 +02:00
@classmethod
async def create_from_preset(
cls,
session: AsyncSession,
diary_id: int,
name: Optional[str],
preset: Preset,
) -> "Meal":
# check if order already exists in diary
query = (
select(cls.order).where(cls.diary_id == diary_id).order_by(cls.order.desc())
)
2023-10-27 15:12:48 +02:00
existing_meal = await session.scalar(query)
order = existing_meal + 1 if existing_meal else 1
2023-10-27 15:12:48 +02:00
if name is None:
name = preset.name or f"Meal {order}"
meal = Meal(diary_id=diary_id, name=name, order=order)
session.add(meal)
try:
await session.flush()
except IntegrityError:
raise AssertionError("diary does not exist")
for entry in preset.entries:
await Entry.create(session, meal.id, entry.product_id, entry.grams)
meal = await cls._get_by_id(session, meal.id)
if not meal:
raise RuntimeError()
return meal
2023-04-01 16:19:12 +02:00
@classmethod
2023-04-03 13:46:03 +02:00
async def _get_by_id(cls, session: AsyncSession, id: int) -> "Optional[Meal]":
2023-04-01 16:19:12 +02:00
"""get_by_id."""
query = select(cls).where(cls.id == id).options(joinedload(cls.entries))
return await session.scalar(query.order_by(cls.id))
2023-04-03 13:46:03 +02:00
@classmethod
async def get_by_id(
cls, session: AsyncSession, user_id: int, id: int
) -> "Optional[Meal]":
"""get_by_id."""
from .diary import Diary
query = (
select(cls)
.where(cls.id == id)
.join(Diary)
.where(Diary.user_id == user_id)
.options(joinedload(cls.entries))
)
return await session.scalar(query.order_by(cls.id))
2023-10-27 16:39:26 +02:00
async def delete(self, session: AsyncSession) -> None:
"""delete."""
for entry in self.entries:
await session.delete(entry)
await session.delete(self)
await session.flush()