111 lines
4.3 KiB
Python
111 lines
4.3 KiB
Python
from contextlib import asynccontextmanager
|
|
from uuid import UUID
|
|
|
|
from fastapi import Depends, FastAPI, Query, Request, status
|
|
from fastapi.exceptions import RequestValidationError
|
|
from fastapi.responses import JSONResponse
|
|
from sqlalchemy import text
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .config import settings
|
|
from .db import get_db, init_db
|
|
from .models import TaskStatus
|
|
from .repository import create_task, delete_task, get_task, list_tasks, update_task
|
|
from .schemas import ErrorResponse, TaskCreate, TaskRead, TaskUpdate
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
init_db()
|
|
yield
|
|
|
|
|
|
app = FastAPI(title=settings.app_name, version="1.0.0", lifespan=lifespan)
|
|
|
|
|
|
def error_response(code: str, message: str, details: list | None = None) -> JSONResponse:
|
|
return JSONResponse(status_code=404 if code == "TASK_NOT_FOUND" else 500, content={"error": {"code": code, "message": message, "details": details or []}})
|
|
|
|
|
|
@app.exception_handler(RequestValidationError)
|
|
async def validation_exception_handler(request: Request, exc: RequestValidationError):
|
|
details = [{"loc": list(error.get("loc", [])), "msg": error.get("msg", "Invalid value"), "type": error.get("type")} for error in exc.errors()]
|
|
return JSONResponse(status_code=422, content={"error": {"code": "VALIDATION_ERROR", "message": "Request validation failed", "details": details}})
|
|
|
|
|
|
@app.get("/health")
|
|
def health(db: Session = Depends(get_db)):
|
|
try:
|
|
db.execute(text("SELECT 1"))
|
|
return {"status": "ok"}
|
|
except SQLAlchemyError:
|
|
return JSONResponse(status_code=503, content={"error": {"code": "DATABASE_UNAVAILABLE", "message": "Database is unavailable", "details": []}})
|
|
|
|
|
|
@app.post("/tasks", response_model=TaskRead, status_code=status.HTTP_201_CREATED, responses={422: {"model": ErrorResponse}})
|
|
def create(data: TaskCreate, db: Session = Depends(get_db)):
|
|
try:
|
|
return create_task(db, data)
|
|
except SQLAlchemyError:
|
|
db.rollback()
|
|
return error_response("DATABASE_ERROR", "Unable to create task")
|
|
|
|
|
|
@app.get("/tasks", response_model=list[TaskRead])
|
|
def list_all(status_filter: TaskStatus | None = Query(default=None, alias="status"), skip: int = Query(0, ge=0), limit: int = Query(100, ge=1, le=100)):
|
|
# dependency is intentionally declared below to keep query parameters obvious in OpenAPI
|
|
return []
|
|
|
|
|
|
@app.get("/tasks", response_model=list[TaskRead], include_in_schema=False)
|
|
def _list_shadow(db: Session = Depends(get_db)):
|
|
return []
|
|
|
|
# Replace the parameter-only route above with the database-backed implementation.
|
|
app.routes.pop(-2)
|
|
|
|
@app.get("/tasks", response_model=list[TaskRead])
|
|
def list_tasks_endpoint(status_filter: TaskStatus | None = Query(default=None, alias="status"), skip: int = Query(0, ge=0), limit: int = Query(100, ge=1, le=100), db: Session = Depends(get_db)):
|
|
return list_tasks(db, status_filter, skip, limit)
|
|
|
|
|
|
def require_task(task_id: UUID, db: Session) :
|
|
task = get_task(db, task_id)
|
|
if task is None:
|
|
raise ValueError("TASK_NOT_FOUND")
|
|
return task
|
|
|
|
|
|
@app.get("/tasks/{task_id}", response_model=TaskRead, responses={404: {"model": ErrorResponse}})
|
|
def get_one(task_id: UUID, db: Session = Depends(get_db)):
|
|
task = get_task(db, task_id)
|
|
if task is None:
|
|
return error_response("TASK_NOT_FOUND", "Task not found")
|
|
return task
|
|
|
|
|
|
@app.patch("/tasks/{task_id}", response_model=TaskRead, responses={404: {"model": ErrorResponse}})
|
|
def patch(task_id: UUID, data: TaskUpdate, db: Session = Depends(get_db)):
|
|
task = get_task(db, task_id)
|
|
if task is None:
|
|
return error_response("TASK_NOT_FOUND", "Task not found")
|
|
try:
|
|
return update_task(db, task, data)
|
|
except SQLAlchemyError:
|
|
db.rollback()
|
|
return error_response("DATABASE_ERROR", "Unable to update task")
|
|
|
|
|
|
@app.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT, responses={404: {"model": ErrorResponse}})
|
|
def remove(task_id: UUID, db: Session = Depends(get_db)):
|
|
task = get_task(db, task_id)
|
|
if task is None:
|
|
return error_response("TASK_NOT_FOUND", "Task not found")
|
|
try:
|
|
delete_task(db, task)
|
|
return None
|
|
except SQLAlchemyError:
|
|
db.rollback()
|
|
return error_response("DATABASE_ERROR", "Unable to delete task")
|