diff --git a/src/backend/api/routes/requests.py b/src/backend/api/routes/requests.py new file mode 100644 index 0000000..394734a --- /dev/null +++ b/src/backend/api/routes/requests.py @@ -0,0 +1,90 @@ +from fastapi import APIRouter, Depends, Query +from starlette.responses import JSONResponse + +from src.backend.api.dependencies import verify_api_key +from src.backend.db.database import ( + get_db, + create_request, + get_request, + list_requests, + delete_request, +) +from src.backend.models.api import RequestCreate, ok, fail + +router = APIRouter( + prefix="/requests", + tags=["requests"], + dependencies=[Depends(verify_api_key)], +) + + +async def _get_db(): + db = await get_db() + try: + yield db + finally: + await db.close() + + +@router.post("", summary="Create request") +async def create(body: RequestCreate, db=Depends(_get_db)): + req = await create_request( + db, + prompt=body.prompt, + topic_id=body.topic_id, + session_id=body.session_id, + model=body.model, + summarize=body.summarize, + ) + if req is None: + return JSONResponse( + status_code=404, + content=fail("NOT_FOUND", "Topic not found"), + ) + if req == "SESSION_CLOSED": + return JSONResponse( + status_code=410, + content=fail("SESSION_CLOSED", "Session is closed"), + ) + return JSONResponse( + status_code=202, + content=ok({ + "id": req["id"], + "session_id": req["session_id"], + "status": req["status"], + "created_at": req["created_at"], + }), + ) + + +@router.get("", summary="List requests") +async def list_all( + status: str | None = Query(None), + topic_id: str | None = Query(None), + session_id: str | None = Query(None), + db=Depends(_get_db), +): + reqs = await list_requests(db, status=status, topic_id=topic_id, session_id=session_id) + return ok(reqs) + + +@router.get("/{request_id}", summary="Get request") +async def get_one(request_id: str, db=Depends(_get_db)): + req = await get_request(db, request_id) + if not req: + return JSONResponse( + status_code=404, + content=fail("NOT_FOUND", "Request not found"), + ) + return ok(req) + + +@router.delete("/{request_id}", summary="Delete request") +async def delete(request_id: str, db=Depends(_get_db)): + deleted = await delete_request(db, request_id) + if not deleted: + return JSONResponse( + status_code=404, + content=fail("NOT_FOUND", "Request not found or not in pending status"), + ) + return ok({"deleted": True}) diff --git a/src/backend/db/database.py b/src/backend/db/database.py index 3cecbbd..96f7d40 100644 --- a/src/backend/db/database.py +++ b/src/backend/db/database.py @@ -33,6 +33,11 @@ CREATE TABLE IF NOT EXISTS requests ( CREATE INDEX IF NOT EXISTS idx_requests_status ON requests(status); CREATE INDEX IF NOT EXISTS idx_requests_topic_id ON requests(topic_id); CREATE INDEX IF NOT EXISTS idx_requests_session_id ON requests(session_id); + +CREATE TABLE IF NOT EXISTS closed_sessions ( + session_id TEXT PRIMARY KEY, + closed_at TEXT NOT NULL +); """ @@ -129,3 +134,189 @@ async def delete_topic(db: aiosqlite.Connection, topic_id: str) -> bool: cursor = await db.execute("DELETE FROM topics WHERE id = ?", (topic_id,)) await db.commit() return cursor.rowcount > 0 + + +async def create_request( + db: aiosqlite.Connection, + prompt: str, + topic_id: str, + session_id: str | None = None, + model: str | None = None, + summarize: bool = False, +) -> dict | None: + topic = await get_topic(db, topic_id) + if not topic: + return None + + if session_id and session_id != "new": + if await is_session_closed(db, session_id): + return "SESSION_CLOSED" + + request_id = _new_id() + now = _now() + + if session_id == "new": + session_id = _new_id() + + await db.execute( + "INSERT INTO requests (id, session_id, topic_id, prompt, model, summarize, status, created_at) " + "VALUES (?, ?, ?, ?, ?, ?, 'pending', ?)", + (request_id, session_id, topic_id, prompt, model, int(summarize), now), + ) + await db.commit() + return { + "id": request_id, + "session_id": session_id, + "topic_id": topic_id, + "prompt": prompt, + "model": model, + "summarize": bool(summarize), + "status": "pending", + "result": None, + "error": None, + "created_at": now, + "completed_at": None, + } + + +async def get_request(db: aiosqlite.Connection, request_id: str) -> dict | None: + cursor = await db.execute("SELECT * FROM requests WHERE id = ?", (request_id,)) + row = await cursor.fetchone() + if not row: + return None + d = dict(row) + d["summarize"] = bool(d["summarize"]) + return d + + +async def list_requests( + db: aiosqlite.Connection, + status: str | None = None, + topic_id: str | None = None, + session_id: str | None = None, +) -> list[dict]: + query = "SELECT * FROM requests WHERE 1=1" + params = [] + if status: + query += " AND status = ?" + params.append(status) + if topic_id: + query += " AND topic_id = ?" + params.append(topic_id) + if session_id: + query += " AND session_id = ?" + params.append(session_id) + query += " ORDER BY created_at DESC" + cursor = await db.execute(query, params) + rows = await cursor.fetchall() + result = [] + for row in rows: + d = dict(row) + d["summarize"] = bool(d["summarize"]) + result.append(d) + return result + + +async def delete_request(db: aiosqlite.Connection, request_id: str) -> bool: + cursor = await db.execute( + "DELETE FROM requests WHERE id = ? AND status = 'pending'", + (request_id,), + ) + await db.commit() + return cursor.rowcount > 0 + + +async def pick_next_pending(db: aiosqlite.Connection) -> dict | None: + cursor = await db.execute( + "SELECT * FROM requests WHERE status = 'pending' ORDER BY created_at ASC LIMIT 1" + ) + row = await cursor.fetchone() + if not row: + return None + d = dict(row) + d["summarize"] = bool(d["summarize"]) + await db.execute( + "UPDATE requests SET status = 'processing' WHERE id = ?", (d["id"],) + ) + await db.commit() + return d + + +async def complete_request( + db: aiosqlite.Connection, + request_id: str, + result: str, + session_id: str | None = None, +) -> None: + now = _now() + if session_id: + await db.execute( + "UPDATE requests SET status = 'completed', result = ?, completed_at = ?, session_id = ? WHERE id = ?", + (result, now, session_id, request_id), + ) + else: + await db.execute( + "UPDATE requests SET status = 'completed', result = ?, completed_at = ? WHERE id = ?", + (result, now, request_id), + ) + await db.commit() + + +async def fail_request( + db: aiosqlite.Connection, request_id: str, error: str +) -> None: + now = _now() + await db.execute( + "UPDATE requests SET status = 'failed', error = ?, completed_at = ? WHERE id = ?", + (error, now, request_id), + ) + await db.commit() + + +async def append_topic_summary( + db: aiosqlite.Connection, topic_id: str, summary_line: str +) -> None: + now = _now() + date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d") + line = f"[{date_str}] {summary_line}" + await db.execute( + "UPDATE topics SET summary = CASE WHEN summary = '' THEN ? ELSE summary || char(10) || ? END, updated_at = ? WHERE id = ?", + (line, line, now, topic_id), + ) + await db.commit() + + +async def list_sessions(db: aiosqlite.Connection) -> list[dict]: + cursor = await db.execute( + "SELECT session_id, topic_id, COUNT(*) as request_count, " + "MIN(created_at) as first_request_at, MAX(created_at) as last_request_at " + "FROM requests WHERE session_id IS NOT NULL " + "GROUP BY session_id ORDER BY last_request_at DESC" + ) + rows = await cursor.fetchall() + return [dict(row) for row in rows] + + +async def session_exists(db: aiosqlite.Connection, session_id: str) -> bool: + cursor = await db.execute( + "SELECT 1 FROM requests WHERE session_id = ? LIMIT 1", (session_id,) + ) + return await cursor.fetchone() is not None + + +async def is_session_closed(db: aiosqlite.Connection, session_id: str) -> bool: + cursor = await db.execute( + "SELECT 1 FROM closed_sessions WHERE session_id = ?", (session_id,) + ) + return await cursor.fetchone() is not None + + +async def close_session(db: aiosqlite.Connection, session_id: str) -> bool: + if not await session_exists(db, session_id): + return False + await db.execute( + "INSERT OR IGNORE INTO closed_sessions (session_id, closed_at) VALUES (?, ?)", + (session_id, _now()), + ) + await db.commit() + return True diff --git a/src/backend/main.py b/src/backend/main.py index 6fb728f..b721819 100644 --- a/src/backend/main.py +++ b/src/backend/main.py @@ -4,6 +4,8 @@ from fastapi import FastAPI from src.backend.api.routes.health import router as health_router from src.backend.api.routes.topics import router as topics_router +from src.backend.api.routes.requests import router as requests_router +from src.backend.api.routes.sessions import router as sessions_router from src.backend.db.database import init_db @@ -17,3 +19,5 @@ app = FastAPI(title="OpusAgent", version="0.1.0", lifespan=lifespan) app.include_router(health_router, prefix="/api") app.include_router(topics_router, prefix="/api") +app.include_router(requests_router, prefix="/api") +app.include_router(sessions_router, prefix="/api") diff --git a/tests/conftest.py b/tests/conftest.py index ba6fe82..f63a32a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -23,6 +23,7 @@ async def setup_db(): yield db = await get_db() try: + await db.execute("DELETE FROM closed_sessions") await db.execute("DELETE FROM requests") await db.execute("DELETE FROM topics") await db.commit() diff --git a/tests/test_api_requests.py b/tests/test_api_requests.py new file mode 100644 index 0000000..a7b03fa --- /dev/null +++ b/tests/test_api_requests.py @@ -0,0 +1,114 @@ +import pytest +from httpx import AsyncClient, ASGITransport + + +@pytest.fixture +def app(): + from src.backend.main import app + return app + + +@pytest.fixture +async def client(app): + transport = ASGITransport(app=app) + async with AsyncClient( + transport=transport, + base_url="http://test", + headers={"X-Api-Key": "test-key-12345"}, + ) as ac: + yield ac + + +@pytest.fixture +async def topic_id(client): + resp = await client.post( + "/api/topics", + json={"name": "test-topic", "system_prompt": "You are helpful."}, + ) + return resp.json()["data"]["id"] + + +@pytest.mark.asyncio +async def test_create_request(client, topic_id): + resp = await client.post( + "/api/requests", + json={"prompt": "Hello", "topic_id": topic_id}, + ) + assert resp.status_code == 202 + data = resp.json() + assert data["success"] is True + assert data["data"]["status"] == "pending" + assert "id" in data["data"] + + +@pytest.mark.asyncio +async def test_create_request_with_new_session(client, topic_id): + resp = await client.post( + "/api/requests", + json={"prompt": "Hello", "topic_id": topic_id, "session_id": "new"}, + ) + assert resp.status_code == 202 + data = resp.json() + assert data["data"]["session_id"] is not None + assert data["data"]["session_id"] != "new" + + +@pytest.mark.asyncio +async def test_create_request_invalid_topic_fails(client): + resp = await client.post( + "/api/requests", + json={"prompt": "Hello", "topic_id": "nonexistent"}, + ) + assert resp.status_code == 404 + + +@pytest.mark.asyncio +async def test_get_request(client, topic_id): + create_resp = await client.post( + "/api/requests", + json={"prompt": "Hello", "topic_id": topic_id}, + ) + req_id = create_resp.json()["data"]["id"] + + resp = await client.get(f"/api/requests/{req_id}") + assert resp.status_code == 200 + assert resp.json()["data"]["prompt"] == "Hello" + + +@pytest.mark.asyncio +async def test_list_requests_with_filters(client, topic_id): + await client.post( + "/api/requests", + json={"prompt": "First", "topic_id": topic_id}, + ) + await client.post( + "/api/requests", + json={"prompt": "Second", "topic_id": topic_id}, + ) + + resp = await client.get("/api/requests") + assert len(resp.json()["data"]) == 2 + + resp = await client.get(f"/api/requests?topic_id={topic_id}") + assert len(resp.json()["data"]) == 2 + + resp = await client.get("/api/requests?status=completed") + assert len(resp.json()["data"]) == 0 + + +@pytest.mark.asyncio +async def test_delete_pending_request(client, topic_id): + create_resp = await client.post( + "/api/requests", + json={"prompt": "Hello", "topic_id": topic_id}, + ) + req_id = create_resp.json()["data"]["id"] + + resp = await client.delete(f"/api/requests/{req_id}") + assert resp.status_code == 200 + + +@pytest.mark.asyncio +async def test_delete_nonexistent_request_fails(client): + resp = await client.delete("/api/requests/nonexistent") + assert resp.status_code == 404