40 lines
1.1 KiB
Python
40 lines
1.1 KiB
Python
from typing import List, Optional
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select
|
|
from app.models.task import Task
|
|
from app.schemas.task import TaskCreate, TaskUpdate
|
|
from app.crud.base import CRUDBase
|
|
|
|
|
|
class CRUDTask(CRUDBase[Task]):
|
|
async def get_by_schedule(
|
|
self,
|
|
db: AsyncSession,
|
|
schedule_id: str
|
|
) -> List[Task]:
|
|
"""Получить все задачи расписания"""
|
|
result = await db.execute(
|
|
select(Task)
|
|
.where(Task.schedule_id == schedule_id)
|
|
.order_by(Task.order)
|
|
)
|
|
return result.scalars().all()
|
|
|
|
async def update_completion(
|
|
self,
|
|
db: AsyncSession,
|
|
task_id: str,
|
|
completed: bool
|
|
) -> Optional[Task]:
|
|
"""Обновить статус выполнения задачи"""
|
|
task = await self.get(db, task_id)
|
|
if task:
|
|
task.completed = completed
|
|
await db.commit()
|
|
await db.refresh(task)
|
|
return task
|
|
|
|
|
|
task = CRUDTask(Task)
|
|
|