aeab17a8ea
- Set up project structure - Created Task model with SQLAlchemy - Implemented SQLite database connection - Created Alembic migrations - Added Task CRUD endpoints - Added health endpoint - Updated README with project details generated with BackendIM... (backend.im)
27 lines
899 B
Python
27 lines
899 B
Python
from sqlalchemy.orm import Session
|
|
from typing import List, Optional
|
|
|
|
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]):
|
|
def get_by_status(self, db: Session, *, status: TaskStatus) -> List[Task]:
|
|
return db.query(self.model).filter(Task.status == status).all()
|
|
|
|
def get_completed(self, db: Session) -> List[Task]:
|
|
return db.query(self.model).filter(Task.completed == True).all()
|
|
|
|
def mark_completed(self, db: Session, *, task_id: int) -> Optional[Task]:
|
|
task = self.get(db, id=task_id)
|
|
if not task:
|
|
return None
|
|
task_in = TaskUpdate(
|
|
status=TaskStatus.DONE,
|
|
completed=True
|
|
)
|
|
return self.update(db, db_obj=task, obj_in=task_in)
|
|
|
|
|
|
task = CRUDTask(Task) |