Piotr Domański
e2af49046c
All checks were successful
Python lint and test / linttest (push) Successful in 5m30s
119 lines
3 KiB
Python
119 lines
3 KiB
Python
import datetime
|
|
from typing import Optional
|
|
|
|
from sqlalchemy import Date, ForeignKey, Integer, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import Mapped, joinedload, mapped_column, relationship
|
|
from sqlalchemy.sql.selectable import Select
|
|
|
|
from .base import Base, CommonMixin
|
|
from .entry import Entry
|
|
from .meal import Meal
|
|
|
|
|
|
class Diary(Base, CommonMixin):
|
|
"""Diary represents user diary for given day"""
|
|
|
|
meals: Mapped[list[Meal]] = relationship(
|
|
lazy="selectin", order_by=Meal.order.desc()
|
|
)
|
|
date: Mapped[datetime.date] = mapped_column(Date)
|
|
user_id: Mapped[int] = mapped_column(Integer, ForeignKey("user.id"))
|
|
|
|
@property
|
|
def calories(self) -> float:
|
|
"""calories.
|
|
|
|
:rtype: float
|
|
"""
|
|
return sum(meal.calories for meal in self.meals)
|
|
|
|
@property
|
|
def protein(self) -> float:
|
|
"""protein.
|
|
|
|
:rtype: float
|
|
"""
|
|
return sum(meal.protein for meal in self.meals)
|
|
|
|
@property
|
|
def carb(self) -> float:
|
|
"""carb.
|
|
|
|
:rtype: float
|
|
"""
|
|
return sum(meal.carb for meal in self.meals)
|
|
|
|
@property
|
|
def fat(self) -> float:
|
|
"""fat.
|
|
|
|
:rtype: float
|
|
"""
|
|
return sum(meal.fat for meal in self.meals)
|
|
|
|
@property
|
|
def fiber(self) -> float:
|
|
"""fiber.
|
|
|
|
:rtype: float
|
|
"""
|
|
return sum(meal.fat for meal in self.meals)
|
|
|
|
@classmethod
|
|
def query(cls, user_id: int) -> Select:
|
|
"""get_all."""
|
|
query = (
|
|
select(cls)
|
|
.where(cls.user_id == user_id)
|
|
.options(
|
|
joinedload(cls.meals).joinedload(Meal.entries).joinedload(Entry.product)
|
|
)
|
|
)
|
|
return query
|
|
|
|
@classmethod
|
|
async def get_diary(
|
|
cls, session: AsyncSession, user_id: int, date: datetime.date
|
|
) -> "Optional[Diary]":
|
|
"""get_diary."""
|
|
query = cls.query(user_id).where(cls.date == date)
|
|
return await session.scalar(query)
|
|
|
|
@classmethod
|
|
async def create(
|
|
cls, session: AsyncSession, user_id: int, date: datetime.date
|
|
) -> "Diary":
|
|
diary = Diary(
|
|
date=date,
|
|
user_id=user_id,
|
|
)
|
|
session.add(diary)
|
|
|
|
try:
|
|
await session.flush()
|
|
except Exception:
|
|
raise RuntimeError()
|
|
|
|
db_diary = await cls.get_by_id(session, user_id, diary.id)
|
|
|
|
if not db_diary:
|
|
raise RuntimeError()
|
|
|
|
await Meal.create(session, db_diary.id)
|
|
return db_diary
|
|
|
|
@classmethod
|
|
async def get_by_id(
|
|
cls, session: AsyncSession, user_id: int, id: int
|
|
) -> "Optional[Diary]":
|
|
"""get_by_id."""
|
|
query = cls.query(user_id).where(cls.id == id)
|
|
return await session.scalar(query)
|
|
|
|
@classmethod
|
|
async def has_permission(cls, session: AsyncSession, user_id: int, id: int) -> bool:
|
|
"""has_permission."""
|
|
query = select(cls.id).where(cls.user_id == user_id).where(cls.id == id)
|
|
obj = await session.scalar(query)
|
|
return obj is not None
|