Eshwar Vanama Spec driven app API

README.md

@@ -41,3 +41,28 @@ Your solution should include at least one real workflow, for example:
 - When you are complete, put up a Pull Request against this repository with your changes.
 - A short summary of your approach and tools used in your PR submission
 - Any additional information or approach that helped you.
+
+# Task Manager API
+
+A simple backend application built using a spec-driven development approach.
+The API supports creating, listing (with filtering and search), updating, and deleting tasks, with data persisted locally using SQLite.
+
+The project demonstrates:
+
+Clear API specification (SPEC.md)
+
+Clean FastAPI + SQLAlchemy architecture
+
+Automated test coverage with pytest
+
+Local development and execution
+## Tech Stack
+- Python 3.11
+- FastAPI
+- SQLite
+- SQLAlchemy
+- Pytest
+
+## Run the application
+```bash
+python -m uvicorn app.main:app --reload

app/crud.py

@@ -0,0 +1,56 @@
+# CRUD operations
+from sqlalchemy.orm import Session
+from sqlalchemy import select
+from .models import Task
+
+
+def create_task(db: Session, title: str, description: str | None):
+    task = Task(
+        title=title,
+        description=description,
+        status="todo"
+    )
+    db.add(task)
+    db.commit()
+    db.refresh(task)
+    return task
+
+
+def get_task(db: Session, task_id: int):
+    return db.get(Task, task_id)
+
+
+def list_tasks(db: Session, status: str | None = None, query: str | None = None):
+    stmt = select(Task)
+
+    if status:
+        stmt = stmt.where(Task.status == status)
+
+    if query:
+        q = f"%{query}%"
+        stmt = stmt.where(
+            Task.title.ilike(q) |
+            Task.description.ilike(q)
+        )
+
+    stmt = stmt.order_by(Task.created_at.desc())
+    return db.scalars(stmt).all()
+
+
+def update_task_status(db: Session, task_id: int, status: str):
+    task = db.get(Task, task_id)
+    if not task:
+        return None
+    task.status = status
+    db.commit()
+    db.refresh(task)
+    return task
+
+
+def delete_task(db: Session, task_id: int):
+    task = db.get(Task, task_id)
+    if not task:
+        return False
+    db.delete(task)
+    db.commit()
+    return True

app/db.py

@@ -0,0 +1,28 @@
+# Database setup and connection
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker, DeclarativeBase
+
+DATABASE_URL = "sqlite:///./db.sqlite3"
+
+engine = create_engine(
+    DATABASE_URL,
+    connect_args={"check_same_thread": False}
+)
+
+SessionLocal = sessionmaker(
+    autocommit=False,
+    autoflush=False,
+    bind=engine
+)
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

app/main.py

@@ -0,0 +1,57 @@
+# Application entry point
+from fastapi import FastAPI, Depends, HTTPException, status
+from sqlalchemy.orm import Session
+
+from .db import Base, engine, get_db
+from . import crud, schemas
+
+Base.metadata.create_all(bind=engine)
+
+app = FastAPI(
+    title="Task Manager API",
+    version="1.0.0"
+)
+
+
+@app.post("/tasks", response_model=schemas.TaskOut, status_code=status.HTTP_201_CREATED)
+def create_task(payload: schemas.TaskCreate, db: Session = Depends(get_db)):
+    if not payload.title.strip():
+        raise HTTPException(status_code=422, detail="title cannot be blank")
+    return crud.create_task(db, payload.title, payload.description)
+
+
+@app.get("/tasks", response_model=list[schemas.TaskOut])
+def list_tasks(
+    status: schemas.TaskStatus | None = None,
+    query: str | None = None,
+    db: Session = Depends(get_db)
+):
+    return crud.list_tasks(db, status=status, query=query)
+
+
+@app.get("/tasks/{task_id}", response_model=schemas.TaskOut)
+def get_task(task_id: int, db: Session = Depends(get_db)):
+    task = crud.get_task(db, task_id)
+    if not task:
+        raise HTTPException(status_code=404, detail="task not found")
+    return task
+
+
+@app.patch("/tasks/{task_id}", response_model=schemas.TaskOut)
+def update_task_status(
+    task_id: int,
+    payload: schemas.TaskUpdateStatus,
+    db: Session = Depends(get_db)
+):
+    task = crud.update_task_status(db, task_id, payload.status)
+    if not task:
+        raise HTTPException(status_code=404, detail="task not found")
+    return task
+
+
+@app.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
+def delete_task(task_id: int, db: Session = Depends(get_db)):
+    ok = crud.delete_task(db, task_id)
+    if not ok:
+        raise HTTPException(status_code=404, detail="task not found")
+    return None

app/models.py

@@ -0,0 +1,21 @@
+# Data models
+from datetime import datetime
+from sqlalchemy import String, Text, DateTime
+from sqlalchemy.orm import Mapped, mapped_column
+from .db import Base
+
+
+class Task(Base):
+    __tablename__ = "tasks"
+
+    id: Mapped[int] = mapped_column(primary_key=True, index=True)
+    title: Mapped[str] = mapped_column(String(200), nullable=False)
+    description: Mapped[str | None] = mapped_column(Text, nullable=True)
+    status: Mapped[str] = mapped_column(String(20), default="todo", nullable=False)
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime, default=datetime.utcnow
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime, default=datetime.utcnow, onupdate=datetime.utcnow
+    )

app/schemas.py

@@ -0,0 +1,32 @@
+# Pydantic schemas
+from datetime import datetime
+from pydantic import BaseModel, Field
+from typing import Optional, Literal
+
+TaskStatus = Literal["todo", "in_progress", "done"]
+
+
+class TaskCreate(BaseModel):
+    title: str = Field(min_length=1, max_length=200)
+    description: Optional[str] = Field(default=None, max_length=2000)
+
+    def model_post_init(self, __context):
+        self.title = self.title.strip()
+        if self.description is not None:
+            self.description = self.description.strip()
+
+
+class TaskUpdateStatus(BaseModel):
+    status: TaskStatus
+
+
+class TaskOut(BaseModel):
+    id: int
+    title: str
+    description: Optional[str]
+    status: TaskStatus
+    created_at: datetime
+    updated_at: datetime
+
+    class Config:
+        from_attributes = True

tests/test_tasks.py

@@ -0,0 +1,110 @@
+# Tests for tasks
+import os
+import tempfile
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from app.db import Base, get_db
+from app.main import app
+
+
+@pytest.fixture()
+def client():
+    # create temp sqlite db for tests
+    fd, path = tempfile.mkstemp(suffix=".sqlite3")
+    os.close(fd)
+
+    engine = create_engine(
+        f"sqlite:///{path}",
+        connect_args={"check_same_thread": False},
+    )
+    TestingSessionLocal = sessionmaker(
+        autocommit=False,
+        autoflush=False,
+        bind=engine,
+    )
+
+    Base.metadata.create_all(bind=engine)
+
+    def override_get_db():
+        db = TestingSessionLocal()
+        try:
+            yield db
+        finally:
+            db.close()
+
+    app.dependency_overrides[get_db] = override_get_db
+
+    with TestClient(app) as client:
+        yield client
+
+    app.dependency_overrides.clear()
+
+    engine.dispose()   # <-- THIS LINE FIXES WINDOWS SQLITE LOCKING
+
+    os.remove(path)
+
+
+def test_create_and_get_task(client):
+    res = client.post(
+        "/tasks",
+        json={"title": "Buy milk", "description": "2 liters"},
+    )
+    assert res.status_code == 201
+
+    task = res.json()
+    assert task["title"] == "Buy milk"
+    assert task["status"] == "todo"
+
+    task_id = task["id"]
+
+    res = client.get(f"/tasks/{task_id}")
+    assert res.status_code == 200
+    assert res.json()["id"] == task_id
+
+
+def test_list_and_search_tasks(client):
+    client.post("/tasks", json={"title": "Alpha"})
+    client.post("/tasks", json={"title": "Beta"})
+    client.post("/tasks", json={"title": "Gamma"})
+
+    res = client.get("/tasks", params={"query": "bet"})
+    assert res.status_code == 200
+
+    tasks = res.json()
+    assert len(tasks) == 1
+    assert tasks[0]["title"] == "Beta"
+
+
+def test_update_status(client):
+    res = client.post("/tasks", json={"title": "Test task"})
+    task_id = res.json()["id"]
+
+    res = client.patch(
+        f"/tasks/{task_id}",
+        json={"status": "in_progress"},
+    )
+    assert res.status_code == 200
+    assert res.json()["status"] == "in_progress"
+
+
+def test_delete_task(client):
+    res = client.post("/tasks", json={"title": "Delete me"})
+    task_id = res.json()["id"]
+
+    res = client.delete(f"/tasks/{task_id}")
+    assert res.status_code == 204
+
+    res = client.get(f"/tasks/{task_id}")
+    assert res.status_code == 404
+
+
+def test_404_cases(client):
+    assert client.get("/tasks/999").status_code == 404
+    assert client.patch(
+        "/tasks/999",
+        json={"status": "done"},
+    ).status_code == 404
+    assert client.delete("/tasks/999").status_code == 404