53bc4e0199
- Set up FastAPI application structure - Create Task model with SQLAlchemy - Implement CRUD operations for tasks - Add API endpoints for tasks with filtering options - Configure Alembic for database migrations - Add health check endpoint - Configure Ruff for linting - Add comprehensive documentation in README.md
28 lines
1013 B
Python
28 lines
1013 B
Python
"""Task CRUD operations."""
|
|
from typing import List
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.crud.base import CRUDBase
|
|
from app.models.task import Task
|
|
from app.schemas.task import TaskCreate, TaskUpdate
|
|
|
|
|
|
class CRUDTask(CRUDBase[Task, TaskCreate, TaskUpdate]):
|
|
"""CRUD operations for tasks."""
|
|
def get_by_title(self, db: Session, *, title: str) -> List[Task]:
|
|
"""Get tasks by title (partial match)."""
|
|
return db.query(Task).filter(Task.title.like(f"%{title}%")).all()
|
|
|
|
def get_completed(self, db: Session, *, skip: int = 0, limit: int = 100) -> List[Task]:
|
|
"""Get completed tasks."""
|
|
return db.query(Task).filter(Task.completed == True).offset(skip).limit(limit).all() # noqa: E712
|
|
|
|
def get_by_priority(
|
|
self, db: Session, *, priority: int, skip: int = 0, limit: int = 100
|
|
) -> List[Task]:
|
|
"""Get tasks by priority."""
|
|
return db.query(Task).filter(Task.priority == priority).offset(skip).limit(limit).all()
|
|
|
|
|
|
task = CRUDTask(Task) |