c9baaa994e
- Add user authentication with JWT tokens - Implement task CRUD operations with status and priority - Set up SQLite database with SQLAlchemy ORM - Create Alembic migrations for database schema - Add comprehensive API documentation - Include health check endpoint and CORS configuration - Structure codebase with proper separation of concerns
80 lines
2.3 KiB
Python
80 lines
2.3 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
from typing import List
|
|
from app.db.session import get_db
|
|
from app.schemas.task import Task, TaskCreate, TaskUpdate
|
|
from app.services.task import (
|
|
get_tasks_by_user,
|
|
get_task_by_id,
|
|
create_task,
|
|
update_task,
|
|
delete_task,
|
|
get_tasks_by_status,
|
|
)
|
|
from app.core.auth import get_current_active_user
|
|
from app.models.user import User
|
|
from app.models.task import TaskStatus
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/", response_model=List[Task])
|
|
def read_tasks(
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
status: TaskStatus = None,
|
|
current_user: User = Depends(get_current_active_user),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
if status:
|
|
return get_tasks_by_status(db, user_id=current_user.id, status=status)
|
|
return get_tasks_by_user(db, user_id=current_user.id, skip=skip, limit=limit)
|
|
|
|
|
|
@router.post("/", response_model=Task)
|
|
def create_new_task(
|
|
task: TaskCreate,
|
|
current_user: User = Depends(get_current_active_user),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
return create_task(db, task=task, user_id=current_user.id)
|
|
|
|
|
|
@router.get("/{task_id}", response_model=Task)
|
|
def read_task(
|
|
task_id: int,
|
|
current_user: User = Depends(get_current_active_user),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
db_task = get_task_by_id(db, task_id=task_id, user_id=current_user.id)
|
|
if db_task is None:
|
|
raise HTTPException(status_code=404, detail="Task not found")
|
|
return db_task
|
|
|
|
|
|
@router.put("/{task_id}", response_model=Task)
|
|
def update_existing_task(
|
|
task_id: int,
|
|
task: TaskUpdate,
|
|
current_user: User = Depends(get_current_active_user),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
db_task = update_task(
|
|
db, task_id=task_id, task_update=task, user_id=current_user.id
|
|
)
|
|
if db_task is None:
|
|
raise HTTPException(status_code=404, detail="Task not found")
|
|
return db_task
|
|
|
|
|
|
@router.delete("/{task_id}")
|
|
def delete_existing_task(
|
|
task_id: int,
|
|
current_user: User = Depends(get_current_active_user),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
success = delete_task(db, task_id=task_id, user_id=current_user.id)
|
|
if not success:
|
|
raise HTTPException(status_code=404, detail="Task not found")
|
|
return {"message": "Task deleted successfully"}
|