29d75cdf08
- Set up project structure and dependencies - Create database models and schemas for tasks - Implement CRUD operations for tasks - Add API endpoints for task management - Create Alembic migrations for the database - Add health check endpoint - Implement error handling - Add documentation in README.md
92 lines
2.3 KiB
Python
92 lines
2.3 KiB
Python
from typing import List, Optional
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.crud.base import CRUDBase
|
|
from app.models.task import Task, TaskStatus
|
|
from app.schemas.task import TaskCreate, TaskUpdate
|
|
|
|
|
|
class CRUDTask(CRUDBase[Task, TaskCreate, TaskUpdate]):
|
|
"""CRUD operations for tasks."""
|
|
|
|
def get_multi_by_status(
|
|
self, db: Session, *, status: TaskStatus, skip: int = 0, limit: int = 100
|
|
) -> List[Task]:
|
|
"""Get multiple tasks by status.
|
|
|
|
Args:
|
|
db: Database session
|
|
status: Task status
|
|
skip: Number of records to skip
|
|
limit: Maximum number of records to return
|
|
|
|
Returns:
|
|
List of tasks with specified status
|
|
"""
|
|
return (
|
|
db.query(self.model)
|
|
.filter(self.model.status == status, not self.model.is_deleted)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
def get_multi(
|
|
self, db: Session, *, skip: int = 0, limit: int = 100
|
|
) -> List[Task]:
|
|
"""Get multiple tasks, excluding deleted ones.
|
|
|
|
Args:
|
|
db: Database session
|
|
skip: Number of records to skip
|
|
limit: Maximum number of records to return
|
|
|
|
Returns:
|
|
List of non-deleted tasks
|
|
"""
|
|
return (
|
|
db.query(self.model)
|
|
.filter(not self.model.is_deleted)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
def get(self, db: Session, id: int) -> Optional[Task]:
|
|
"""Get a task by ID, excluding deleted ones.
|
|
|
|
Args:
|
|
db: Database session
|
|
id: Task ID
|
|
|
|
Returns:
|
|
The task if found and not deleted, None otherwise
|
|
"""
|
|
return (
|
|
db.query(self.model)
|
|
.filter(self.model.id == id, not self.model.is_deleted)
|
|
.first()
|
|
)
|
|
|
|
def remove(self, db: Session, *, id: int) -> Task:
|
|
"""Soft delete a task.
|
|
|
|
Args:
|
|
db: Database session
|
|
id: Task ID
|
|
|
|
Returns:
|
|
The soft deleted task
|
|
"""
|
|
obj = db.query(self.model).get(id)
|
|
if obj:
|
|
obj.is_deleted = True
|
|
db.add(obj)
|
|
db.commit()
|
|
db.refresh(obj)
|
|
return obj
|
|
|
|
|
|
task = CRUDTask(Task)
|