From 23b0162a06f1eeee75b180aa2121f4353f5bbdf3 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Wed, 7 Oct 2026 11:39:49 -0700 Subject: [PATCH] feat: sort param on /recent and /search; microsecond last_used_at - /prompt_manager/recent and /search accept ?sort= (whitelisted in the DB layer; unknown values fall back to newest first). - last_used_at is written with microseconds from Python so a re-run in the same millisecond as another save still sorts first; ordering stays correct against millisecond values from the backfill. Refs #90 --- database/models.py | 13 +++++++-- database/operations.py | 11 +++---- py/api/prompts.py | 8 ++++-- tests/test_api.py | 59 ++++++++++++++++++++++++++++++++++++++ tests/test_prompt_usage.py | 20 +++++++++++++ 5 files changed, 102 insertions(+), 9 deletions(-) diff --git a/database/models.py b/database/models.py index e38f7ba..5dc2cfe 100644 --- a/database/models.py +++ b/database/models.py @@ -2,6 +2,7 @@ Database schema and models for KikoTextEncode prompt storage. """ +import datetime import sqlite3 import os import threading @@ -17,9 +18,17 @@ except ImportError: sys.path.insert(0, current_dir) from utils.logging_config import get_logger -# Single UTC format for last_used_at so ORDER BY can compare it as text +# last_used_at is UTC ISO text so ORDER BY can compare it as text. The backfill +# writes milliseconds (SQLite's %f); live writes use microseconds so runs in the +# same millisecond still order correctly. Both forms sort correctly together. ISO_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%f+00:00" -SQL_NOW_ISO = f"strftime('{ISO_TIMESTAMP_FORMAT}', 'now')" + + +def utc_now_iso() -> str: + """Current UTC time as ISO text with microseconds, e.g. 2026-10-07T18:39:06.758123+00:00.""" + return datetime.datetime.now(datetime.timezone.utc).isoformat( + timespec="microseconds" + ) class PromptModel: diff --git a/database/operations.py b/database/operations.py index 52c40cd..5a82cfd 100644 --- a/database/operations.py +++ b/database/operations.py @@ -8,7 +8,7 @@ import datetime import os from typing import Optional, List, Dict, Any, Union -from .models import PromptModel, SQL_NOW_ISO +from .models import PromptModel, utc_now_iso # Import logging system try: @@ -139,11 +139,11 @@ class PromptDatabase: with self.model.get_connection() as conn: cursor = conn.execute( - f""" + """ INSERT INTO prompts ( text, category, tags, rating, notes, hash, created_at, updated_at, last_used_at - ) VALUES (?, ?, NULL, ?, ?, ?, ?, ?, {SQL_NOW_ISO}) + ) VALUES (?, ?, NULL, ?, ?, ?, ?, ?, ?) """, ( text.strip(), @@ -153,6 +153,7 @@ class PromptDatabase: prompt_hash, datetime.datetime.now(datetime.timezone.utc).isoformat(), datetime.datetime.now(datetime.timezone.utc).isoformat(), + utc_now_iso(), ), ) prompt_id = cursor.lastrowid @@ -174,9 +175,9 @@ class PromptDatabase: """ with self.model.get_connection() as conn: cursor = conn.execute( - f"UPDATE prompts SET run_count = run_count + 1, last_used_at = {SQL_NOW_ISO}" + "UPDATE prompts SET run_count = run_count + 1, last_used_at = ?" " WHERE id = ?", - (prompt_id,), + (utc_now_iso(), prompt_id), ) conn.commit() return cursor.rowcount > 0 diff --git a/py/api/prompts.py b/py/api/prompts.py index 4a47eed..5d0efdf 100644 --- a/py/api/prompts.py +++ b/py/api/prompts.py @@ -132,6 +132,8 @@ class PromptRoutesMixin: limit = int(request.query.get("limit", 50)) folder = request.query.get("folder", "").strip() or None + # Validated against a whitelist in the database layer + sort = request.query.get("sort") or None tags = None if tags_str: @@ -150,6 +152,7 @@ class PromptRoutesMixin: rating_min=min_rating, limit=limit, folder=folder, + sort=sort, ) self._enrich_prompt_images(results) @@ -182,7 +185,7 @@ class PromptRoutesMixin: return web.json_response({"success": False, "error": str(e)}, status=500) async def get_recent_prompts(self, request): - """Retrieve recently created prompts with pagination support.""" + """Retrieve prompts with pagination and an optional sort (default newest first).""" try: limit = int(request.query.get("limit", 50)) page = int(request.query.get("page", 1)) @@ -196,8 +199,9 @@ class PromptRoutesMixin: elif limit < 1: limit = 1 + sort = request.query.get("sort") or None results = await self._run_in_executor( - self.db.get_recent_prompts, limit=limit, offset=offset + self.db.get_recent_prompts, limit=limit, offset=offset, sort=sort ) self._enrich_prompt_images(results["prompts"]) diff --git a/tests/test_api.py b/tests/test_api.py index f985353..d7bc416 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -238,6 +238,65 @@ class TestUpdatePrompt(APITestCase): self.assertTrue(data["success"]) +class TestUsageSorting(APITestCase): + """Server-side sort for /recent and /search (3.2.4, #90.1).""" + + def _use(self, prompt_id, times=1): + for _ in range(times): + self.api.db.record_prompt_use(prompt_id) + + async def _ids(self, path): + resp = await self.client.request("GET", path) + self.assertEqual(resp.status, 200) + data = await resp.json() + self.assertTrue(data["success"]) + return [p["id"] for p in data["results"]] + + async def test_recent_sorted_by_last_used_bubbles_rerun_prompt(self): + old = self._save_prompt("Old prompt") + self._save_prompt("New prompt") + self._use(old) + ids = await self._ids("/prompt_manager/recent?limit=1&sort=last_used_desc") + self.assertEqual(ids, [old]) + + async def test_recent_sorted_by_run_count(self): + # Most-run prompt is the oldest, so newest-first would give the opposite order + most = self._save_prompt("Run thrice") + least = self._save_prompt("Run once") + self._use(most, times=3) + self._use(least) + ids = await self._ids("/prompt_manager/recent?sort=run_count_desc") + self.assertEqual(ids, [most, least]) + + async def test_recent_default_is_newest_first(self): + first = self._save_prompt("First") + second = self._save_prompt("Second") + self._use(first) + self.assertEqual(await self._ids("/prompt_manager/recent"), [second, first]) + + async def test_unknown_sort_falls_back_instead_of_erroring(self): + self._save_prompt("Only") + ids = await self._ids( + "/prompt_manager/recent?sort=created_at;DROP%20TABLE%20prompts" + ) + self.assertEqual(len(ids), 1) + + async def test_search_respects_sort(self): + a = self._save_prompt("castle at dawn") + self._save_prompt("castle at dusk") + self._use(a) + ids = await self._ids("/prompt_manager/search?text=castle&sort=last_used_desc") + self.assertEqual(ids[0], a) + + async def test_results_include_usage_fields(self): + pid = self._save_prompt("Tracked") + self._use(pid, times=2) + resp = await self.client.request("GET", "/prompt_manager/recent") + prompt = (await resp.json())["results"][0] + self.assertEqual(prompt["run_count"], 2) + self.assertTrue(prompt["last_used_at"]) + + class TestTags(APITestCase): async def test_get_all_tags(self): diff --git a/tests/test_prompt_usage.py b/tests/test_prompt_usage.py index 1d9e6ed..2980f7d 100644 --- a/tests/test_prompt_usage.py +++ b/tests/test_prompt_usage.py @@ -57,6 +57,26 @@ class TestRecordPromptUse(UsageTestCase): self.assertEqual(prompt["run_count"], 2) self.assertGreater(prompt["last_used_at"], "2020-01-01T00:00:00.000+00:00") + def test_rerun_right_after_another_save_still_sorts_first(self): + # No hand-set timestamps: creation and re-run happen within the same millisecond + old = self._save("old") + self._save("new") + self.db.record_prompt_use(old) + result = self.db.get_recent_prompts(limit=1, sort="last_used_desc") + self.assertEqual(self._ids(result["prompts"]), [old]) + + def test_microsecond_timestamps_order_correctly_against_backfilled_ones(self): + backfilled = self._save("backfilled") + self._set(backfilled, last_used_at="2026-01-01T00:00:00.000+00:00") + recent = self._save("recent") + self.db.record_prompt_use(recent) + self.assertRegex( + self.db.get_prompt_by_id(recent)["last_used_at"], + r"^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d\.\d{6}\+00:00$", + ) + result = self.db.get_recent_prompts(limit=2, sort="last_used_desc") + self.assertEqual(self._ids(result["prompts"]), [recent, backfilled]) + def test_record_use_on_missing_prompt_returns_false(self): self.assertFalse(self.db.record_prompt_use(999999))