decomposer: generate deliverable files for Implement the task REST API service with FastAPI, including standard CRUD endpoints, HTTP status semantics, request and response models, and structured error handling.; Add a PostgreSQL persistence layer for the task service using SQLAlchemy models, session management, database initialization or migrations, and durable task storage integrated with the existing API contract.; Define and integrate Pydantic request and response schemas for task fields, status values, identifiers, timestamps, and validation errors across the existing FastAPI service and persistence contract.; Add automated pytest coverage for the task API, including CRUD behavior, request and response validation failures, HTTP status semantics, and persistence interactions using isolated test data.; Containerize the FastAPI task service and PostgreSQL persistence service with Docker Compose, including environment-based configuration, health checks, networking, startup dependencies, and persistent database storage.; Create project documentation and operational guidance covering local setup, Docker Compose configuration, environment variables, API endpoints, validation and error responses, testing commands, persistence behavior, and relevant best-practice notes.

app/config.py

@@ -1 +1,11 @@
-"""Typed settings."""
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    app_name: str = "Task API"
+    database_url: str = "postgresql+psycopg2://tasks:tasks@localhost:5432/tasks"
+    log_level: str = "INFO"
+    model_config = SettingsConfigDict(env_file=".env", extra="ignore")
+
+
+settings = Settings()

app/db.py

@@ -0,0 +1,26 @@
+from collections.abc import Generator
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
+
+from .config import settings
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+engine = create_engine(settings.database_url, pool_pre_ping=True)
+SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, expire_on_commit=False)
+
+
+def init_db() -> None:
+    Base.metadata.create_all(bind=engine)
+
+
+def get_db() -> Generator[Session, None, None]:
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

app/main.py

@@ -0,0 +1,110 @@
+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")

app/models.py

@@ -0,0 +1,26 @@
+import enum
+import uuid
+from datetime import datetime
+
+from sqlalchemy import DateTime, Enum, String, Text, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from .db import Base
+
+
+class TaskStatus(str, enum.Enum):
+    pending = "pending"
+    in_progress = "in_progress"
+    completed = "completed"
+
+
+class Task(Base):
+    __tablename__ = "tasks"
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
+    title: Mapped[str] = mapped_column(String(200), nullable=False, index=True)
+    description: Mapped[str | None] = mapped_column(Text, nullable=True)
+    status: Mapped[TaskStatus] = mapped_column(Enum(TaskStatus, name="task_status"), nullable=False, default=TaskStatus.pending)
+    due_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), nullable=False)
+    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False)

app/repository.py

@@ -0,0 +1,39 @@
+from uuid import UUID
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from .models import Task, TaskStatus
+from .schemas import TaskCreate, TaskUpdate
+
+
+def create_task(db: Session, data: TaskCreate) -> Task:
+    task = Task(**data.model_dump())
+    db.add(task)
+    db.commit()
+    db.refresh(task)
+    return task
+
+
+def list_tasks(db: Session, status: TaskStatus | None, skip: int, limit: int) -> list[Task]:
+    query = select(Task).order_by(Task.created_at.desc()).offset(skip).limit(limit)
+    if status is not None:
+        query = query.where(Task.status == status)
+    return list(db.scalars(query).all())
+
+
+def get_task(db: Session, task_id: UUID) -> Task | None:
+    return db.get(Task, str(task_id))
+
+
+def update_task(db: Session, task: Task, data: TaskUpdate) -> Task:
+    for key, value in data.model_dump(exclude_unset=True).items():
+        setattr(task, key, value)
+    db.commit()
+    db.refresh(task)
+    return task
+
+
+def delete_task(db: Session, task: Task) -> None:
+    db.delete(task)
+    db.commit()

app/schemas.py

@@ -0,0 +1,64 @@
+from datetime import datetime
+from uuid import UUID
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+from .models import TaskStatus
+
+
+class TaskBase(BaseModel):
+    title: str = Field(min_length=1, max_length=200)
+    description: str | None = Field(default=None, max_length=5000)
+    status: TaskStatus = TaskStatus.pending
+    due_at: datetime | None = None
+
+    @field_validator("title")
+    @classmethod
+    def title_not_blank(cls, value: str) -> str:
+        value = value.strip()
+        if not value:
+            raise ValueError("title must not be blank")
+        return value
+
+
+class TaskCreate(TaskBase):
+    pass
+
+
+class TaskUpdate(BaseModel):
+    title: str | None = Field(default=None, min_length=1, max_length=200)
+    description: str | None = Field(default=None, max_length=5000)
+    status: TaskStatus | None = None
+    due_at: datetime | None = None
+
+    @field_validator("title")
+    @classmethod
+    def title_not_blank(cls, value: str | None) -> str | None:
+        if value is not None:
+            value = value.strip()
+            if not value:
+                raise ValueError("title must not be blank")
+        return value
+
+
+class TaskRead(TaskBase):
+    id: UUID
+    created_at: datetime
+    updated_at: datetime
+    model_config = ConfigDict(from_attributes=True)
+
+
+class ErrorDetail(BaseModel):
+    loc: list[str | int] = []
+    msg: str
+    type: str | None = None
+
+
+class ErrorBody(BaseModel):
+    code: str
+    message: str
+    details: list[ErrorDetail] = []
+
+
+class ErrorResponse(BaseModel):
+    error: ErrorBody