refactor: DRY pass on backend — pagination helper and sharing utilities

- Add parse_pagination() to routes/utils.py; replace 6 duplicate limit/offset extractions in notes, tasks, chat, projects, milestones routes
- Extract _enrich_shares() in sharing.py; eliminates identical 12-line loop in list_project_shares and list_note_shares
- Extract _deduplicate_by_permission() in sharing.py; eliminates identical deduplication blocks in list_shared_with_me for projects and notes

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

src/fabledassistant/routes/chat.py

@@ -6,7 +6,7 @@ import httpx
 from quart import Blueprint, Response, jsonify, request
 
 from fabledassistant.auth import login_required, get_current_user_id
-from fabledassistant.routes.utils import not_found
+from fabledassistant.routes.utils import not_found, parse_pagination
 from fabledassistant.config import Config
 from fabledassistant.services.chat import (
     add_message,
@@ -38,8 +38,7 @@ chat_bp = Blueprint("chat", __name__, url_prefix="/api/chat")
 @login_required
 async def list_conversations_route():
     uid = get_current_user_id()
-    limit = min(request.args.get("limit", 50, type=int), 500)
-    offset = request.args.get("offset", 0, type=int)
+    limit, offset = parse_pagination()
     conv_type = request.args.get("type", "chat")
     # Apply retention policy before returning list
     retention_str = await get_setting(uid, "chat_retention_days", "90")

src/fabledassistant/routes/milestones.py

@@ -4,7 +4,7 @@ import logging
 from quart import Blueprint, jsonify, request
 
 from fabledassistant.auth import login_required, get_current_user_id
-from fabledassistant.routes.utils import not_found
+from fabledassistant.routes.utils import not_found, parse_pagination
 from fabledassistant.services.milestones import (
     create_milestone,
     delete_milestone,
@@ -107,8 +107,7 @@ async def get_milestone_tasks_route(project_id: int, milestone_id: int):
     if milestone is None or milestone.project_id != project_id:
         return not_found("Milestone")
     status_filter = request.args.get("status")
-    limit = min(request.args.get("limit", 100, type=int), 500)
-    offset = request.args.get("offset", 0, type=int)
+    limit, offset = parse_pagination(default_limit=100)
     notes, total = await list_notes(
         uid,
         is_task=True,

src/fabledassistant/routes/notes.py

@@ -7,7 +7,7 @@ from fabledassistant.services.embeddings import upsert_note_embedding
 from quart import Blueprint, Response, jsonify, request
 
 from fabledassistant.auth import login_required, get_current_user_id
-from fabledassistant.routes.utils import not_found, parse_iso_date
+from fabledassistant.routes.utils import not_found, parse_iso_date, parse_pagination
 from fabledassistant.config import Config
 from fabledassistant.services.assist import build_assist_messages
 from fabledassistant.services.generation_buffer import (
@@ -49,8 +49,7 @@ async def list_notes_route():
     tag = request.args.getlist("tag")
     sort = request.args.get("sort", "updated_at")
     order = request.args.get("order", "desc")
-    limit = min(request.args.get("limit", 50, type=int), 500)
-    offset = request.args.get("offset", 0, type=int)
+    limit, offset = parse_pagination()
 
     # Default to non-task notes only; ?is_task=true for tasks, ?all=true for everything
     is_task: bool | None = False

src/fabledassistant/routes/projects.py

@@ -4,7 +4,7 @@ import logging
 from quart import Blueprint, jsonify, request
 
 from fabledassistant.auth import login_required, get_current_user_id
-from fabledassistant.routes.utils import not_found
+from fabledassistant.routes.utils import not_found, parse_pagination
 from fabledassistant.services.milestones import list_milestones
 from fabledassistant.services.notes import list_notes
 from fabledassistant.services.projects import (
@@ -100,8 +100,7 @@ async def get_project_notes_route(project_id: int):
     # type filter: "note", "task", or None (both)
     type_filter = request.args.get("type")
     status_filter = request.args.get("status")
-    limit = min(request.args.get("limit", 100, type=int), 500)
-    offset = request.args.get("offset", 0, type=int)
+    limit, offset = parse_pagination(default_limit=100)
 
     is_task: bool | None = None
     if type_filter == "task":

src/fabledassistant/routes/tasks.py

@@ -4,7 +4,7 @@ from quart import Blueprint, jsonify, request
 
 from fabledassistant.auth import login_required, get_current_user_id
 from fabledassistant.models.note import TaskPriority, TaskStatus
-from fabledassistant.routes.utils import not_found, parse_iso_date
+from fabledassistant.routes.utils import not_found, parse_iso_date, parse_pagination
 from fabledassistant.services.embeddings import upsert_note_embedding
 from fabledassistant.services.notes import (
     create_note,
@@ -28,8 +28,7 @@ async def list_tasks_route():
     priority = request.args.get("priority")
     sort = request.args.get("sort", "updated_at")
     order = request.args.get("order", "desc")
-    limit = min(request.args.get("limit", 50, type=int), 500)
-    offset = request.args.get("offset", 0, type=int)
+    limit, offset = parse_pagination()
 
     project_id = request.args.get("project_id", type=int)
     no_project = request.args.get("no_project", "").lower() == "true"

src/fabledassistant/routes/utils.py

@@ -1,6 +1,6 @@
 from datetime import date
 
-from quart import jsonify
+from quart import jsonify, request
 
 
 def not_found(resource: str = "Item"):
@@ -15,3 +15,10 @@ def parse_iso_date(value: str | None, field: str = "date"):
         return date.fromisoformat(value)
     except ValueError:
         return jsonify({"error": f"Invalid {field} format. Use YYYY-MM-DD."}), 400
+
+
+def parse_pagination(default_limit: int = 50, max_limit: int = 500) -> tuple[int, int]:
+    """Extract and clamp ``limit`` / ``offset`` from the current request's query string."""
+    limit = min(request.args.get("limit", default_limit, type=int), max_limit)
+    offset = request.args.get("offset", 0, type=int)
+    return limit, offset

src/fabledassistant/services/sharing.py

@@ -16,6 +16,33 @@ logger = logging.getLogger(__name__)
 VALID_PERMISSIONS = {"viewer", "editor", "admin"}
 
 
+async def _enrich_shares(session, shares) -> list[dict]:
+    """Attach username / group_name to a list of share rows (ProjectShare or NoteShare)."""
+    result = []
+    for s in shares:
+        d = s.to_dict()
+        if s.shared_with_user_id:
+            user = await session.get(User, s.shared_with_user_id)
+            d["username"] = user.username if user else None
+        if s.shared_with_group_id:
+            group = await session.get(Group, s.shared_with_group_id)
+            d["group_name"] = group.name if group else None
+        result.append(d)
+    return result
+
+
+def _deduplicate_by_permission(shares, id_attr: str) -> dict[int, str]:
+    """Return {resource_id: best_permission} keeping the highest-ranked permission per resource."""
+    from fabledassistant.services.access import PERMISSION_RANK
+    seen: dict[int, str] = {}
+    for share in shares:
+        rid = getattr(share, id_attr)
+        prev = seen.get(rid)
+        if prev is None or PERMISSION_RANK[share.permission] > PERMISSION_RANK[prev]:
+            seen[rid] = share.permission
+    return seen
+
+
 # ---------------------------------------------------------------------------
 # Project shares
 # ---------------------------------------------------------------------------
@@ -83,17 +110,7 @@ async def list_project_shares(project_id: int) -> list[dict]:
         shares = (await session.execute(
             select(ProjectShare).where(ProjectShare.project_id == project_id)
         )).scalars().all()
-        result = []
-        for s in shares:
-            d = s.to_dict()
-            if s.shared_with_user_id:
-                user = await session.get(User, s.shared_with_user_id)
-                d["username"] = user.username if user else None
-            if s.shared_with_group_id:
-                group = await session.get(Group, s.shared_with_group_id)
-                d["group_name"] = group.name if group else None
-            result.append(d)
-        return result
+        return await _enrich_shares(session, shares)
 
 
 # ---------------------------------------------------------------------------
@@ -166,17 +183,7 @@ async def list_note_shares(note_id: int) -> list[dict]:
         shares = (await session.execute(
             select(NoteShare).where(NoteShare.note_id == note_id)
         )).scalars().all()
-        result = []
-        for s in shares:
-            d = s.to_dict()
-            if s.shared_with_user_id:
-                user = await session.get(User, s.shared_with_user_id)
-                d["username"] = user.username if user else None
-            if s.shared_with_group_id:
-                group = await session.get(Group, s.shared_with_group_id)
-                d["group_name"] = group.name if group else None
-            result.append(d)
-        return result
+        return await _enrich_shares(session, shares)
 
 
 # ---------------------------------------------------------------------------
@@ -203,12 +210,7 @@ async def list_shared_with_me(user_id: int) -> dict:
                 )
             )).scalars().all()
 
-        seen_projects: dict[int, str] = {}
-        for share in list(proj_direct) + list(proj_group):
-            prev = seen_projects.get(share.project_id)
-            from fabledassistant.services.access import PERMISSION_RANK
-            if prev is None or PERMISSION_RANK[share.permission] > PERMISSION_RANK[prev]:
-                seen_projects[share.project_id] = share.permission
+        seen_projects = _deduplicate_by_permission(list(proj_direct) + list(proj_group), "project_id")
 
         projects = []
         for pid, perm in seen_projects.items():
@@ -239,12 +241,7 @@ async def list_shared_with_me(user_id: int) -> dict:
                 )
             )).scalars().all()
 
-        seen_notes: dict[int, str] = {}
-        for share in list(note_direct) + list(note_group):
-            prev = seen_notes.get(share.note_id)
-            from fabledassistant.services.access import PERMISSION_RANK
-            if prev is None or PERMISSION_RANK[share.permission] > PERMISSION_RANK[prev]:
-                seen_notes[share.note_id] = share.permission
+        seen_notes = _deduplicate_by_permission(list(note_direct) + list(note_group), "note_id")
 
         notes = []
         for nid, perm in seen_notes.items():