28f6914da1
- Set up project structure with FastAPI - Implement SQLAlchemy models for User and Task - Create Alembic migrations - Implement authentication with JWT - Add CRUD operations for tasks - Add task filtering and prioritization - Configure health check endpoint - Update README with project documentation
38 lines
1.1 KiB
Python
38 lines
1.1 KiB
Python
from typing import List
|
|
|
|
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]):
|
|
def create_with_owner(self, db: Session, *, obj_in: TaskCreate, owner_id: str) -> Task:
|
|
return super().create(db, obj_in=obj_in, owner_id=owner_id)
|
|
|
|
def get_multi_by_owner(
|
|
self, db: Session, *, owner_id: str, skip: int = 0, limit: int = 100
|
|
) -> List[Task]:
|
|
return (
|
|
db.query(self.model)
|
|
.filter(Task.owner_id == owner_id)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
def get_by_status(
|
|
self, db: Session, *, owner_id: str, status: TaskStatus, skip: int = 0, limit: int = 100
|
|
) -> List[Task]:
|
|
return (
|
|
db.query(self.model)
|
|
.filter(Task.owner_id == owner_id, Task.status == status)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
|
|
task = CRUDTask(Task)
|