40 lines
1.3 KiB
Python
40 lines
1.3 KiB
Python
|
from uuid import UUID
|
||
|
|
|
||
|
|
from sqlalchemy import select
|
||
|
|
from sqlalchemy.orm import selectinload
|
||
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
|
|
||
|
|
from app.crud.base import CRUDBase
|
||
|
|
from app.models.planting import BedPlanting
|
||
|
|
from app.models.plant import Plant
|
||
|
|
from app.schemas.planting import PlantingCreate, PlantingUpdate
|
||
|
|
|
||
|
|
|
||
|
|
class CRUDPlanting(CRUDBase[BedPlanting, PlantingCreate, PlantingUpdate]):
|
||
|
|
async def get(self, db: AsyncSession, *, id: UUID) -> BedPlanting | None:
|
||
|
|
result = await db.execute(
|
||
|
|
select(BedPlanting)
|
||
|
|
.options(selectinload(BedPlanting.plant).selectinload(Plant.family))
|
||
|
|
.where(BedPlanting.id == id)
|
||
|
|
)
|
||
|
|
return result.scalar_one_or_none()
|
||
|
|
|
||
|
|
async def get_multi_for_bed(
|
||
|
|
self, db: AsyncSession, *, bed_id: UUID
|
||
|
|
) -> list[BedPlanting]:
|
||
|
|
result = await db.execute(
|
||
|
|
select(BedPlanting)
|
||
|
|
.options(selectinload(BedPlanting.plant).selectinload(Plant.family))
|
||
|
|
.where(BedPlanting.bed_id == bed_id)
|
||
|
|
.order_by(BedPlanting.planted_date.desc().nullslast())
|
||
|
|
)
|
||
|
|
return list(result.scalars().all())
|
||
|
|
|
||
|
|
async def create_for_bed(
|
||
|
|
self, db: AsyncSession, *, obj_in: PlantingCreate, bed_id: UUID
|
||
|
|
) -> BedPlanting:
|
||
|
|
return await self.create(db, obj_in=obj_in, bed_id=bed_id)
|
||
|
|
|
||
|
|
|
||
|
|
crud_planting = CRUDPlanting(BedPlanting)
|