56 lines
1.8 KiB
Python
56 lines
1.8 KiB
Python
from typing import List, Optional
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from sqlalchemy import select
|
||
from sqlalchemy.orm import selectinload
|
||
from datetime import date
|
||
from app.models.schedule import Schedule
|
||
from app.schemas.schedule import ScheduleCreate, ScheduleUpdate
|
||
from app.crud.base import CRUDBase
|
||
|
||
|
||
class CRUDSchedule(CRUDBase[Schedule]):
|
||
async def get_by_user(
|
||
self,
|
||
db: AsyncSession,
|
||
user_id: str,
|
||
skip: int = 0,
|
||
limit: int = 100
|
||
) -> List[Schedule]:
|
||
"""Получить расписания пользователя"""
|
||
result = await db.execute(
|
||
select(Schedule)
|
||
.where(Schedule.user_id == user_id)
|
||
.options(selectinload(Schedule.tasks))
|
||
.offset(skip)
|
||
.limit(limit)
|
||
.order_by(Schedule.date.desc())
|
||
)
|
||
return result.scalars().all()
|
||
|
||
async def get_by_date(
|
||
self,
|
||
db: AsyncSession,
|
||
user_id: str,
|
||
schedule_date: date
|
||
) -> Optional[Schedule]:
|
||
"""Получить расписание на конкретную дату"""
|
||
result = await db.execute(
|
||
select(Schedule)
|
||
.where(Schedule.user_id == user_id, Schedule.date == schedule_date)
|
||
.options(selectinload(Schedule.tasks))
|
||
)
|
||
return result.scalar_one_or_none()
|
||
|
||
async def get_with_tasks(self, db: AsyncSession, id: str) -> Optional[Schedule]:
|
||
"""Получить расписание с задачами"""
|
||
result = await db.execute(
|
||
select(Schedule)
|
||
.where(Schedule.id == id)
|
||
.options(selectinload(Schedule.tasks))
|
||
)
|
||
return result.scalar_one_or_none()
|
||
|
||
|
||
schedule = CRUDSchedule(Schedule)
|
||
|