40 lines
1.0 KiB
Python
40 lines
1.0 KiB
Python
from uuid import UUID
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .models import Task, TaskStatus
|
|
from .schemas import TaskCreate, TaskUpdate
|
|
|
|
|
|
def create_task(db: Session, data: TaskCreate) -> Task:
|
|
task = Task(**data.model_dump())
|
|
db.add(task)
|
|
db.commit()
|
|
db.refresh(task)
|
|
return task
|
|
|
|
|
|
def list_tasks(db: Session, status: TaskStatus | None, skip: int, limit: int) -> list[Task]:
|
|
query = select(Task).order_by(Task.created_at.desc()).offset(skip).limit(limit)
|
|
if status is not None:
|
|
query = query.where(Task.status == status)
|
|
return list(db.scalars(query).all())
|
|
|
|
|
|
def get_task(db: Session, task_id: UUID) -> Task | None:
|
|
return db.get(Task, str(task_id))
|
|
|
|
|
|
def update_task(db: Session, task: Task, data: TaskUpdate) -> Task:
|
|
for key, value in data.model_dump(exclude_unset=True).items():
|
|
setattr(task, key, value)
|
|
db.commit()
|
|
db.refresh(task)
|
|
return task
|
|
|
|
|
|
def delete_task(db: Session, task: Task) -> None:
|
|
db.delete(task)
|
|
db.commit()
|