fooder-api/fooder/domain/entry.py

123 lines
3.2 KiB
Python
Raw Normal View History

2023-04-01 16:19:12 +02:00
from sqlalchemy.orm import Mapped, mapped_column, relationship, joinedload
from sqlalchemy import ForeignKey, Integer, DateTime
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.exc import IntegrityError
from sqlalchemy import select
from datetime import datetime
from typing import Optional
from .base import Base, CommonMixin
from .product import Product
class Entry(Base, CommonMixin):
"""Entry."""
grams: Mapped[float]
product_id: Mapped[int] = mapped_column(Integer, ForeignKey("product.id"))
product: Mapped[Product] = relationship(lazy="selectin")
meal_id: Mapped[int] = mapped_column(Integer, ForeignKey("meal.id"))
last_changed: Mapped[datetime] = mapped_column(
DateTime, default=datetime.utcnow, onupdate=datetime.utcnow
)
@property
def amount(self) -> float:
"""amount.
:rtype: float
"""
return self.grams / 100
@property
def calories(self) -> float:
"""calories.
:rtype: float
"""
return self.amount * self.product.calories
@property
def protein(self) -> float:
"""protein.
:rtype: float
"""
return self.amount * self.product.protein
@property
def carb(self) -> float:
"""carb.
:rtype: float
"""
return self.amount * self.product.carb
@property
def fat(self) -> float:
"""fat.
:rtype: float
"""
return self.amount * self.product.fat
@classmethod
async def create(
cls, session: AsyncSession, meal_id: int, product_id: int, grams: float
) -> "Entry":
"""create."""
assert grams > 0, "grams must be greater than 0"
entry = Entry(
meal_id=meal_id,
product_id=product_id,
grams=grams,
)
session.add(entry)
try:
await session.flush()
except IntegrityError:
raise AssertionError("meal or product does not exist")
entry = await cls.get_by_id(session, entry.id)
if not entry:
raise RuntimeError()
return entry
async def update(
self,
session: AsyncSession,
meal_id: Optional[int],
product_id: Optional[int],
grams: Optional[float],
) -> None:
"""update."""
if grams is not None:
assert grams > 0, "grams must be greater than 0"
self.grams = grams
if meal_id is not None:
self.meal_id = meal_id
try:
session.flush()
except IntegrityError:
raise AssertionError("meal does not exist")
if product_id is not None:
self.product_id = product_id
try:
session.flush()
except IntegrityError:
raise AssertionError("product does not exist")
@classmethod
async def get_by_id(cls, session: AsyncSession, id: int) -> "Optional[Entry]":
"""get_by_id."""
query = select(cls).where(cls.id == id).options(joinedload(cls.product))
return await session.scalar(query.order_by(cls.id))
async def delete(self, session) -> None:
"""delete."""
await session.delete(self)
await session.flush()