114 lines
4.3 KiB
Python
114 lines
4.3 KiB
Python
from typing import List, Sequence
|
|
|
|
from sqlalchemy import delete, func, select, text
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import contains_eager, selectinload
|
|
|
|
from app.core.database.models import ApproachModel, ExerciseModel, TrainingModel
|
|
from app.core.dto.training import (
|
|
ApproachDTO,
|
|
ExerciseNameDTO,
|
|
ExerciseSessionDTO,
|
|
TrainingDTO,
|
|
)
|
|
|
|
|
|
class TrainingRepository:
|
|
"""Storage for trainings together with their exercises and approaches."""
|
|
|
|
def __init__(self, session: AsyncSession) -> None:
|
|
self._session = session
|
|
|
|
async def save(self, training: TrainingDTO) -> int:
|
|
return (await self.save_many([training]))[0]
|
|
|
|
async def save_many(self, trainings: Sequence[TrainingDTO]) -> List[int]:
|
|
"""Save all trainings in a single transaction, returns their ids."""
|
|
models = [self._to_model(training) for training in trainings]
|
|
self._session.add_all(models)
|
|
await self._session.commit()
|
|
return [model.id for model in models]
|
|
|
|
async def list(self) -> List[TrainingDTO]:
|
|
query = (
|
|
select(TrainingModel)
|
|
.options(
|
|
selectinload(TrainingModel.exercises).selectinload(
|
|
ExerciseModel.approaches
|
|
)
|
|
)
|
|
.order_by(TrainingModel.date.desc(), TrainingModel.id.desc())
|
|
)
|
|
result = await self._session.scalars(query)
|
|
return [TrainingDTO.model_validate(model) for model in result]
|
|
|
|
async def list_exercise_names(self) -> List[ExerciseNameDTO]:
|
|
"""Distinct exercise names with the number of times each was done."""
|
|
query = select(ExerciseModel.name, func.count(ExerciseModel.id)).group_by(
|
|
ExerciseModel.name
|
|
)
|
|
result = await self._session.execute(query)
|
|
names = [ExerciseNameDTO(name=name, count=count) for name, count in result]
|
|
# Sorted here, not in SQL: the order must not depend on DB collation
|
|
return sorted(names, key=lambda item: item.name.casefold())
|
|
|
|
async def exercise_history(self, name: str) -> List[ExerciseSessionDTO]:
|
|
"""Every occurrence of the exercise, newest first."""
|
|
query = (
|
|
select(ExerciseModel)
|
|
.join(ExerciseModel.training)
|
|
.where(ExerciseModel.name == name)
|
|
.options(
|
|
contains_eager(ExerciseModel.training),
|
|
selectinload(ExerciseModel.approaches),
|
|
)
|
|
.order_by(TrainingModel.date.desc(), ExerciseModel.id.desc())
|
|
)
|
|
result = await self._session.scalars(query)
|
|
return [
|
|
ExerciseSessionDTO(
|
|
date=exercise.training.date,
|
|
trainer=exercise.training.trainer,
|
|
splitted_weight=exercise.splitted_weight,
|
|
approaches=[
|
|
ApproachDTO.model_validate(approach)
|
|
for approach in exercise.approaches
|
|
],
|
|
)
|
|
for exercise in result
|
|
]
|
|
|
|
async def count(self) -> int:
|
|
return await self._session.scalar(select(func.count(TrainingModel.id))) or 0
|
|
|
|
async def delete(self, list_of_ids: List[int]) -> int:
|
|
"""Delete trainings by ids (exercises and approaches go with them)."""
|
|
result = await self._session.execute(
|
|
delete(TrainingModel).where(TrainingModel.id.in_(list_of_ids))
|
|
)
|
|
await self._session.commit()
|
|
return result.rowcount # type: ignore[attr-defined]
|
|
|
|
async def delete_all(self) -> None:
|
|
await self._session.execute(
|
|
text("TRUNCATE trainings RESTART IDENTITY CASCADE")
|
|
)
|
|
await self._session.commit()
|
|
|
|
@staticmethod
|
|
def _to_model(training: TrainingDTO) -> TrainingModel:
|
|
return TrainingModel(
|
|
date=training.date,
|
|
trainer=training.trainer,
|
|
exercises=[
|
|
ExerciseModel(
|
|
name=exercise.name,
|
|
splitted_weight=exercise.splitted_weight,
|
|
approaches=[
|
|
ApproachModel(weight=approach.weight, reps=approach.reps)
|
|
for approach in exercise.approaches
|
|
],
|
|
)
|
|
for exercise in training.exercises
|
|
],
|
|
)
|