app/main.py

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+from threading import Lock
+from typing import Optional
+
+from fastapi import FastAPI, HTTPException
+from prometheus_fastapi_instrumentator import Instrumentator
+from pydantic import BaseModel
+
+app = FastAPI(
+    title="sonar-test-pyt",
+    description="sonar-test-pyt",
+    version="0.1.0",
+)
+
+# Expose /metrics for Prometheus scraping
+Instrumentator().instrument(app).expose(app)
+
+
+# ---------------------------------------------------------------------------
+# Domain model
+# ---------------------------------------------------------------------------
+class ItemRequest(BaseModel):
+    name: str
+    description: str = ""
+
+
+class Item(BaseModel):
+    id: int
+    name: str
+    description: str
+
+
+# ---------------------------------------------------------------------------
+# In-memory store (thread-safe)
+# ---------------------------------------------------------------------------
+_store: dict[int, Item] = {}
+_counter = 0
+_lock = Lock()
+
+
+def _next_id() -> int:
+    global _counter
+    with _lock:
+        _counter += 1
+        return _counter
+
+
+# ---------------------------------------------------------------------------
+# Routes
+# ---------------------------------------------------------------------------
+@app.get("/health", tags=["observability"])
+def health() -> dict:
+    return {"status": "UP"}
+
+
+@app.get("/api/items", response_model=list[Item], tags=["items"])
+def list_items():
+    return list(_store.values())
+
+
+@app.post("/api/items", response_model=Item, status_code=201, tags=["items"])
+def create_item(body: ItemRequest):
+    item = Item(id=_next_id(), name=body.name, description=body.description)
+    _store[item.id] = item
+    return item
+
+
+@app.get("/api/items/{item_id}", response_model=Item, tags=["items"])
+def get_item(item_id: int):
+    item = _store.get(item_id)
+    if item is None:
+        raise HTTPException(status_code=404, detail="Item not found")
+    return item
+
+
+@app.put("/api/items/{item_id}", response_model=Item, tags=["items"])
+def update_item(item_id: int, body: ItemRequest):
+    item = _store.get(item_id)
+    if item is None:
+        raise HTTPException(status_code=404, detail="Item not found")
+    updated = Item(id=item_id, name=body.name, description=body.description)
+    _store[item_id] = updated
+    return updated
+
+
+@app.delete("/api/items/{item_id}", tags=["items"])
+def delete_item(item_id: int):
+    _store.pop(item_id, None)
+    return {"deleted": item_id}