320 lines
10 KiB
Python
320 lines
10 KiB
Python
from fastapi import FastAPI, Depends, HTTPException, UploadFile, File
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from sqlalchemy.orm import Session
|
|
from typing import List, Optional
|
|
import json
|
|
|
|
from . import models, schemas, crud
|
|
from .database import engine, get_db
|
|
from .settings import settings
|
|
|
|
# Create database tables
|
|
models.Base.metadata.create_all(bind=engine)
|
|
|
|
app = FastAPI(
|
|
title=settings.api_title,
|
|
description=settings.api_description,
|
|
version=settings.api_version
|
|
)
|
|
|
|
# CORS middleware for frontend
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=settings.cors_origins_list,
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
|
|
# ========== PROJECT ENDPOINTS ==========
|
|
|
|
@app.get("/api/projects", response_model=List[schemas.Project])
|
|
def list_projects(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
|
|
"""List all projects"""
|
|
return crud.get_projects(db, skip=skip, limit=limit)
|
|
|
|
|
|
@app.post("/api/projects", response_model=schemas.Project, status_code=201)
|
|
def create_project(project: schemas.ProjectCreate, db: Session = Depends(get_db)):
|
|
"""Create a new project"""
|
|
return crud.create_project(db, project)
|
|
|
|
|
|
@app.get("/api/projects/{project_id}", response_model=schemas.Project)
|
|
def get_project(project_id: int, db: Session = Depends(get_db)):
|
|
"""Get a specific project"""
|
|
db_project = crud.get_project(db, project_id)
|
|
if not db_project:
|
|
raise HTTPException(status_code=404, detail="Project not found")
|
|
return db_project
|
|
|
|
|
|
@app.put("/api/projects/{project_id}", response_model=schemas.Project)
|
|
def update_project(
|
|
project_id: int, project: schemas.ProjectUpdate, db: Session = Depends(get_db)
|
|
):
|
|
"""Update a project"""
|
|
db_project = crud.update_project(db, project_id, project)
|
|
if not db_project:
|
|
raise HTTPException(status_code=404, detail="Project not found")
|
|
return db_project
|
|
|
|
|
|
@app.delete("/api/projects/{project_id}", status_code=204)
|
|
def delete_project(project_id: int, db: Session = Depends(get_db)):
|
|
"""Delete a project and all its tasks"""
|
|
if not crud.delete_project(db, project_id):
|
|
raise HTTPException(status_code=404, detail="Project not found")
|
|
return None
|
|
|
|
|
|
# ========== TASK ENDPOINTS ==========
|
|
|
|
@app.get("/api/projects/{project_id}/tasks", response_model=List[schemas.Task])
|
|
def list_project_tasks(project_id: int, db: Session = Depends(get_db)):
|
|
"""List all tasks for a project"""
|
|
if not crud.get_project(db, project_id):
|
|
raise HTTPException(status_code=404, detail="Project not found")
|
|
return crud.get_tasks_by_project(db, project_id)
|
|
|
|
|
|
@app.get("/api/projects/{project_id}/tasks/tree", response_model=List[schemas.TaskWithSubtasks])
|
|
def get_project_task_tree(project_id: int, db: Session = Depends(get_db)):
|
|
"""Get the task tree (root tasks with nested subtasks) for a project"""
|
|
if not crud.get_project(db, project_id):
|
|
raise HTTPException(status_code=404, detail="Project not found")
|
|
|
|
root_tasks = crud.get_root_tasks(db, project_id)
|
|
|
|
def build_tree(task):
|
|
task_dict = schemas.TaskWithSubtasks.model_validate(task)
|
|
task_dict.subtasks = [build_tree(subtask) for subtask in task.subtasks]
|
|
return task_dict
|
|
|
|
return [build_tree(task) for task in root_tasks]
|
|
|
|
|
|
@app.get("/api/projects/{project_id}/tasks/by-status/{status}", response_model=List[schemas.Task])
|
|
def get_tasks_by_status(
|
|
project_id: int,
|
|
status: str,
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Get all tasks for a project filtered by status (for Kanban view)"""
|
|
if not crud.get_project(db, project_id):
|
|
raise HTTPException(status_code=404, detail="Project not found")
|
|
try:
|
|
return crud.get_tasks_by_status(db, project_id, status)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
|
|
@app.post("/api/tasks", response_model=schemas.Task, status_code=201)
|
|
def create_task(task: schemas.TaskCreate, db: Session = Depends(get_db)):
|
|
"""Create a new task"""
|
|
if not crud.get_project(db, task.project_id):
|
|
raise HTTPException(status_code=404, detail="Project not found")
|
|
|
|
if task.parent_task_id and not crud.get_task(db, task.parent_task_id):
|
|
raise HTTPException(status_code=404, detail="Parent task not found")
|
|
|
|
try:
|
|
return crud.create_task(db, task)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
|
|
@app.get("/api/tasks/{task_id}", response_model=schemas.Task)
|
|
def get_task(task_id: int, db: Session = Depends(get_db)):
|
|
"""Get a specific task"""
|
|
db_task = crud.get_task(db, task_id)
|
|
if not db_task:
|
|
raise HTTPException(status_code=404, detail="Task not found")
|
|
return db_task
|
|
|
|
|
|
@app.put("/api/tasks/{task_id}", response_model=schemas.Task)
|
|
def update_task(task_id: int, task: schemas.TaskUpdate, db: Session = Depends(get_db)):
|
|
"""Update a task"""
|
|
try:
|
|
db_task = crud.update_task(db, task_id, task)
|
|
if not db_task:
|
|
raise HTTPException(status_code=404, detail="Task not found")
|
|
return db_task
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
|
|
@app.delete("/api/tasks/{task_id}", status_code=204)
|
|
def delete_task(task_id: int, db: Session = Depends(get_db)):
|
|
"""Delete a task and all its subtasks"""
|
|
if not crud.delete_task(db, task_id):
|
|
raise HTTPException(status_code=404, detail="Task not found")
|
|
return None
|
|
|
|
|
|
# ========== SEARCH ENDPOINT ==========
|
|
|
|
@app.get("/api/search", response_model=List[schemas.Task])
|
|
def search_tasks(
|
|
query: str,
|
|
project_ids: Optional[str] = None,
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""
|
|
Search tasks across projects by title, description, and tags.
|
|
|
|
Args:
|
|
query: Search term to match against title, description, and tags
|
|
project_ids: Comma-separated list of project IDs to search in (optional, searches all if not provided)
|
|
"""
|
|
# Parse project IDs if provided
|
|
project_id_list = None
|
|
if project_ids:
|
|
try:
|
|
project_id_list = [int(pid.strip()) for pid in project_ids.split(',') if pid.strip()]
|
|
except ValueError:
|
|
raise HTTPException(status_code=400, detail="Invalid project_ids format")
|
|
|
|
# Build query
|
|
tasks_query = db.query(models.Task)
|
|
|
|
# Filter by project IDs if specified
|
|
if project_id_list:
|
|
tasks_query = tasks_query.filter(models.Task.project_id.in_(project_id_list))
|
|
|
|
# Search in title, description, and tags
|
|
search_term = f"%{query}%"
|
|
tasks = tasks_query.filter(
|
|
(models.Task.title.ilike(search_term)) |
|
|
(models.Task.description.ilike(search_term)) |
|
|
(models.Task.tags.contains([query])) # Exact tag match
|
|
).all()
|
|
|
|
return tasks
|
|
|
|
|
|
# ========== JSON IMPORT ENDPOINT ==========
|
|
|
|
def _validate_task_statuses_recursive(
|
|
tasks: List[schemas.ImportSubtask],
|
|
valid_statuses: List[str],
|
|
path: str = ""
|
|
) -> None:
|
|
"""Recursively validate all task statuses against the project's valid statuses"""
|
|
for idx, task_data in enumerate(tasks):
|
|
task_path = f"{path}.tasks[{idx}]" if path else f"tasks[{idx}]"
|
|
if task_data.status not in valid_statuses:
|
|
raise ValueError(
|
|
f"Invalid status '{task_data.status}' at {task_path}. "
|
|
f"Must be one of: {', '.join(valid_statuses)}"
|
|
)
|
|
if task_data.subtasks:
|
|
_validate_task_statuses_recursive(
|
|
task_data.subtasks,
|
|
valid_statuses,
|
|
f"{task_path}.subtasks"
|
|
)
|
|
|
|
|
|
def _import_tasks_recursive(
|
|
db: Session,
|
|
project_id: int,
|
|
tasks: List[schemas.ImportSubtask],
|
|
parent_id: Optional[int] = None,
|
|
count: int = 0
|
|
) -> int:
|
|
"""Recursively import tasks and their subtasks"""
|
|
for idx, task_data in enumerate(tasks):
|
|
task = schemas.TaskCreate(
|
|
project_id=project_id,
|
|
parent_task_id=parent_id,
|
|
title=task_data.title,
|
|
description=task_data.description,
|
|
status=task_data.status,
|
|
estimated_minutes=task_data.estimated_minutes,
|
|
tags=task_data.tags,
|
|
flag_color=task_data.flag_color,
|
|
sort_order=idx
|
|
)
|
|
db_task = crud.create_task(db, task)
|
|
count += 1
|
|
|
|
if task_data.subtasks:
|
|
count = _import_tasks_recursive(
|
|
db, project_id, task_data.subtasks, db_task.id, count
|
|
)
|
|
|
|
return count
|
|
|
|
|
|
@app.post("/api/import-json", response_model=schemas.ImportResult)
|
|
def import_from_json(import_data: schemas.ImportData, db: Session = Depends(get_db)):
|
|
"""
|
|
Import a project with nested tasks from JSON.
|
|
|
|
Expected format:
|
|
{
|
|
"project": {
|
|
"name": "Project Name",
|
|
"description": "Optional description",
|
|
"statuses": ["backlog", "in_progress", "on_hold", "done"] // Optional
|
|
},
|
|
"tasks": [
|
|
{
|
|
"title": "Task 1",
|
|
"description": "Optional",
|
|
"status": "backlog",
|
|
"subtasks": [
|
|
{
|
|
"title": "Subtask 1.1",
|
|
"status": "backlog",
|
|
"subtasks": []
|
|
}
|
|
]
|
|
}
|
|
]
|
|
}
|
|
"""
|
|
# Create the project with optional statuses
|
|
project = crud.create_project(
|
|
db,
|
|
schemas.ProjectCreate(
|
|
name=import_data.project.name,
|
|
description=import_data.project.description,
|
|
statuses=import_data.project.statuses
|
|
)
|
|
)
|
|
|
|
# Validate all task statuses before importing
|
|
if import_data.tasks:
|
|
try:
|
|
_validate_task_statuses_recursive(import_data.tasks, project.statuses)
|
|
except ValueError as e:
|
|
# Rollback the project creation if validation fails
|
|
db.delete(project)
|
|
db.commit()
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
# Recursively import tasks
|
|
tasks_created = _import_tasks_recursive(
|
|
db, project.id, import_data.tasks
|
|
)
|
|
|
|
return schemas.ImportResult(
|
|
project_id=project.id,
|
|
project_name=project.name,
|
|
tasks_created=tasks_created
|
|
)
|
|
|
|
|
|
@app.get("/")
|
|
def root():
|
|
"""API health check"""
|
|
return {
|
|
"status": "online",
|
|
"message": "Tesseract API - Nested Todo Tree Manager",
|
|
"docs": "/docs"
|
|
}
|