builderadmin
/
NoteDiscovery
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

Merge pull request #187 from gamosoft/features/pagination 

Features/pagination
This commit is contained in:
Guillermo Villar 2026-03-23 18:08:13 +01:00 committed by GitHub
parent f5147ccc14 eb49dad678
commit 19c3259b16
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 418 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

133
backend/main.py
View File

@ -42,6 +42,7 @@ from .utils import (
get_templates,
get_template_content,
apply_template_placeholders,
paginate,
)
from .plugins import PluginManager
from .themes import get_available_themes, get_theme_css
@ -876,23 +877,48 @@ async def list_tags():
@api_router.get("/tags/{tag_name}", tags=["Tags"])
async def get_notes_by_tag_endpoint(tag_name: str):
async def get_notes_by_tag_endpoint(
tag_name: str,
limit: Optional[int] = None,
offset: int = 0
):
"""
Get all notes that have a specific tag.
Get all notes that have a specific tag with optional pagination.
Args:
tag_name: The tag to filter by (case-insensitive)
limit: Maximum number of notes to return (optional, no default limit)
offset: Number of notes to skip (default: 0)
Returns:
List of notes matching the tag
Examples:
GET /api/tags/docker -> All notes with #docker tag
GET /api/tags/docker?limit=10 -> First 10 notes with #docker tag
"""
try:
notes = get_notes_by_tag(config['storage']['notes_dir'], tag_name)
return {
# Apply pagination with consistent sorting by path
paginated = paginate(
items=notes,
limit=limit,
offset=offset,
sort_key=lambda x: x.get('path', '').lower()
)
response = {
"tag": tag_name,
"count": len(notes),
"notes": notes
"count": paginated.total,
"notes": paginated.items
}
# Include pagination metadata only when limit is specified
if limit is not None:
response["pagination"] = paginated.to_dict()
return response
except Exception as e:
raise HTTPException(status_code=500, detail=safe_error_message(e, "Failed to get notes by tag"))
@ -1004,11 +1030,47 @@ async def create_note_from_template(request: Request, data: dict):
# --- Notes Endpoints ---
@api_router.get("/notes", tags=["Notes"])
async def list_notes():
"""List all notes with metadata"""
async def list_notes(
limit: Optional[int] = None,
offset: int = 0
):
"""
List all notes with metadata.
Supports optional pagination for API consumers (MCP, scripts):
- No parameters: Returns all notes (frontend compatibility)
- With limit: Returns paginated results with metadata
Args:
limit: Maximum number of notes to return (optional, no default limit)
offset: Number of notes to skip (default: 0)
Examples:
GET /api/notes -> All notes
GET /api/notes?limit=20 -> First 20 notes
GET /api/notes?limit=20&offset=20 -> Notes 21-40
"""
try:
notes, folders = scan_notes_fast_walk(config['storage']['notes_dir'], include_media=True)
return {"notes": notes, "folders": folders}
# Apply pagination with consistent sorting by path for stable results
result = paginate(
items=notes,
limit=limit,
offset=offset,
sort_key=lambda x: x.get('path', '').lower()
)
response = {
"notes": result.items,
"folders": folders
}
# Include pagination metadata only when limit is specified
if limit is not None:
response["pagination"] = result.to_dict()
return response
except Exception as e:
raise HTTPException(status_code=500, detail=safe_error_message(e, "Failed to list notes"))
@ -1161,18 +1223,59 @@ async def remove_note(request: Request, note_path: str):
@api_router.get("/search", tags=["Search"])
async def search(q: str):
"""Search notes by content"""
async def search(
q: str,
limit: Optional[int] = None,
offset: int = 0
):
"""
Search notes by content with optional pagination.
Args:
q: Search query string
limit: Maximum number of results to return (optional, no default limit)
offset: Number of results to skip (default: 0)
Examples:
GET /api/search?q=docker -> All matching results
GET /api/search?q=docker&limit=10 -> First 10 results
GET /api/search?q=docker&limit=10&offset=10 -> Results 11-20
"""
try:
if not config['search']['enabled']:
raise HTTPException(status_code=403, detail="Search is disabled")
# Handle empty query gracefully
if not q or not q.strip():
return {
"results": [],
"query": q,
"message": "No search term provided"
}
results = search_notes(config['storage']['notes_dir'], q)
# Run plugin hooks
plugin_manager.run_hook('on_search', query=q, results=results)
# Apply pagination with consistent sorting by path
paginated = paginate(
items=results,
limit=limit,
offset=offset,
sort_key=lambda x: x.get('path', '').lower()
)
return {"results": results, "query": q}
response = {
"results": paginated.items,
"query": q
}
# Include pagination metadata only when limit is specified
if limit is not None:
response["pagination"] = paginated.to_dict()
return response
except HTTPException:
raise
except Exception as e:

100
backend/utils.py
View File

@ -7,11 +7,109 @@ import re
import shutil
import threading
import time
from dataclasses import dataclass
from pathlib import Path
from typing import List, Dict, Optional, Tuple
from typing import List, Dict, Optional, Tuple, Any, TypeVar, Callable
from datetime import datetime, timezone
# ============================================================================
# Pagination Support
# ============================================================================
@dataclass
class PaginationResult:
"""
Result of applying pagination to a list.
Attributes:
items: The paginated subset of items
total: Total number of items before pagination
limit: The limit that was applied (None if no pagination)
offset: The offset that was applied
has_more: Whether there are more items after this page
"""
items: List[Any]
total: int
limit: Optional[int]
offset: int
has_more: bool
def to_dict(self) -> Dict[str, Any]:
"""Convert pagination info to dict for API response."""
return {
"limit": self.limit,
"offset": self.offset,
"total": self.total,
"has_more": self.has_more
}
T = TypeVar('T')
def paginate(
items: List[T],
limit: Optional[int] = None,
offset: int = 0,
sort_key: Optional[Callable[[T], Any]] = None,
sort_reverse: bool = False
) -> PaginationResult:
"""
Apply optional pagination to a list with consistent sorting.
This function is designed to be backward-compatible:
- If limit is None, returns all items (no pagination)
- If limit is provided, returns a paginated subset
Sorting is always applied (when sort_key is provided) to ensure
stable pagination across requests.
Args:
items: List of items to paginate
limit: Maximum number of items to return (None = no limit)
offset: Number of items to skip (default: 0)
sort_key: Function to extract sort key from item (e.g., lambda x: x['path'])
sort_reverse: If True, sort in descending order
Returns:
PaginationResult with items and pagination metadata
Example:
# No pagination (frontend compatibility)
result = paginate(notes)
# With pagination (MCP usage)
result = paginate(notes, limit=20, offset=0, sort_key=lambda x: x['path'])
"""
total = len(items)
# Apply sorting for consistent ordering (prevents out-of-order issues)
if sort_key is not None:
items = sorted(items, key=sort_key, reverse=sort_reverse)
# Apply pagination only if limit is specified
if limit is not None:
# Clamp offset to valid range
offset = max(0, min(offset, total))
end = offset + limit
paginated_items = items[offset:end]
has_more = end < total
else:
# No pagination - return all items
paginated_items = items
offset = 0
has_more = False
return PaginationResult(
items=paginated_items,
total=total,
limit=limit,
offset=offset,
has_more=has_more
)
# In-memory cache for parsed tags
# Format: {file_path: (mtime, tags)}
_tag_cache: Dict[str, Tuple[float, List[str]]] = {}

43
documentation/API.md
View File

@ -7,12 +7,28 @@ Base URL: `http://localhost:8000`
### List All Notes
```http
GET /api/notes
GET /api/notes?limit=20&offset=0
```
Returns all notes with their metadata and folder structure.
**Example:**
**Optional Pagination:**
| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `limit` | integer | - | Max notes to return (omit for all) |
| `offset` | integer | 0 | Number of notes to skip |
When `limit` is provided, the response includes pagination metadata.
**Examples:**
```bash
# Get all notes (default)
curl http://localhost:8000/api/notes
# Get first 20 notes
curl "http://localhost:8000/api/notes?limit=20"
# Get notes 21-40
curl "http://localhost:8000/api/notes?limit=20&offset=20"
```
### Get Note Content
@ -269,11 +285,22 @@ Content-Type: application/json
### Search Notes
```http
GET /api/search?q={query}
GET /api/search?q={query}&limit=10&offset=0
```
**Example:**
**Optional Pagination:**
| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `limit` | integer | - | Max results to return (omit for all) |
| `offset` | integer | 0 | Number of results to skip |
**Examples:**
```bash
# Search all matches
curl "http://localhost:8000/api/search?q=hello"
# Get first 10 results
curl "http://localhost:8000/api/search?q=hello&limit=10"
```
## 🎨 Themes
@ -409,14 +436,24 @@ Returns all tags found in notes with their usage counts.
```
### Get Notes by Tag
`GET /api/tags/{tag_name}`
```http
GET /api/tags/{tag_name}
GET /api/tags/{tag_name}?limit=10&offset=0
```
Returns all notes that have a specific tag.
**Optional Pagination:**
| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `limit` | integer | - | Max notes to return (omit for all) |
| `offset` | integer | 0 | Number of notes to skip |
**Response:**
```json
{
"tag": "python",
"count": 5,
"notes": [
{
"path": "tutorials/python-basics.md",

25
documentation/MCP.md
View File

@ -114,8 +114,8 @@ The MCP server provides these tools to AI assistants:
| Tool | Description |
|------|-------------|
| `search_notes` | Full-text search across all notes |
| `list_notes` | List all notes with metadata |
| `search_notes` | Full-text search across all notes (supports `max_results`) |
| `list_notes` | List all notes with metadata (supports `max_results`) |
| `get_note` | Read a specific note's content |
| `get_recent_notes` | Get recently modified notes (last N days) |
@ -124,7 +124,7 @@ The MCP server provides these tools to AI assistants:
| Tool | Description |
|------|-------------|
| `list_tags` | List all tags with note counts |
| `get_notes_by_tag` | Find notes with a specific tag |
| `get_notes_by_tag` | Find notes with a specific tag (supports `max_results`) |
| `get_graph` | Get knowledge graph data |
### Note Management
@ -153,6 +153,25 @@ The MCP server provides these tools to AI assistants:
## Tool Details
### Pagination with `max_results` and `offset`
Some tools support optional pagination parameters for large vaults:
| Tool | Parameters | Description |
|------|------------|-------------|
| `search_notes` | `max_results`, `offset` | Paginate search results |
| `list_notes` | `max_results`, `offset` | Paginate notes list |
| `get_notes_by_tag` | `max_results`, `offset` | Paginate notes by tag |
- `max_results` - Maximum items to return (omit for all)
- `offset` - Number of items to skip (for pagination)
**Example prompts:**
- "Search for notes about Python, but just show me the first 5 results"
- "Show me the next 5 Python notes" (uses offset)
---
### `append_to_note`
Append content to an existing note without overwriting. Perfect for journals, logs, or collecting ideas.

50
mcp_server/client.py
View File

@ -139,14 +139,26 @@ class NoteDiscoveryClient:
# Notes API
# =========================================================================
def list_notes(self) -> APIResponse:
def list_notes(
self,
limit: int | None = None,
offset: int = 0
) -> APIResponse:
"""
List all notes with metadata.
Args:
limit: Maximum number of notes to return (None = no limit)
offset: Number of notes to skip (default: 0)
Returns:
APIResponse with notes and folders data
"""
return self._request("GET", "/api/notes")
params: dict[str, str] = {}
if limit is not None:
params["limit"] = str(limit)
params["offset"] = str(offset)
return self._request("GET", "/api/notes", params=params if params else None)
def get_note(self, path: str) -> APIResponse:
"""
@ -255,17 +267,28 @@ class NoteDiscoveryClient:
# Search API
# =========================================================================
def search(self, query: str) -> APIResponse:
def search(
self,
query: str,
limit: int | None = None,
offset: int = 0
) -> APIResponse:
"""
Search notes by query.
Args:
query: Search query string
limit: Maximum number of results to return (None = no limit)
offset: Number of results to skip (default: 0)
Returns:
APIResponse with search results
"""
return self._request("GET", "/api/search", params={"q": query})
params: dict[str, str] = {"q": query}
if limit is not None:
params["limit"] = str(limit)
params["offset"] = str(offset)
return self._request("GET", "/api/search", params=params)
# =========================================================================
# Tags API
@ -280,18 +303,33 @@ class NoteDiscoveryClient:
"""
return self._request("GET", "/api/tags")
def get_notes_by_tag(self, tag: str) -> APIResponse:
def get_notes_by_tag(
self,
tag: str,
limit: int | None = None,
offset: int = 0
) -> APIResponse:
"""
Get notes with a specific tag.
Args:
tag: Tag name
limit: Maximum number of notes to return (None = no limit)
offset: Number of notes to skip (default: 0)
Returns:
APIResponse with matching notes
"""
encoded_tag = urllib.parse.quote(tag, safe="")
return self._request("GET", f"/api/tags/{encoded_tag}")
params: dict[str, str] = {}
if limit is not None:
params["limit"] = str(limit)
params["offset"] = str(offset)
return self._request(
"GET",
f"/api/tags/{encoded_tag}",
params=params if params else None
)
# =========================================================================
# Folders API

89
mcp_server/server.py
View File

@ -8,6 +8,7 @@ This implementation uses only Python stdlib for minimal dependencies.
"""
import json
import re
import sys
import traceback
from typing import Any, Optional
@ -206,39 +207,62 @@ class MCPServer:
def _tool_search_notes(self, args: dict) -> str:
"""Search notes by query."""
query = args.get("query", "")
if not query:
return "Error: query is required"
query = args.get("query", "").strip()
max_results = args.get("max_results")
offset = args.get("offset", 0)
response = self.client.search(query)
if not query:
return "No search term provided. Please specify a query."
# Use server-side pagination if max_results is specified
response = self.client.search(query, limit=max_results, offset=offset)
if not response.success:
return f"Search failed: {response.error}"
data = response.data or {}
results = data.get("results", [])
pagination = data.get("pagination")
if not results:
return f"No notes found matching '{query}'"
# Build header with pagination info if available
if pagination:
total = pagination.get("total", len(results))
output = [f"Found {total} result(s) for '{query}' (showing {len(results)}):\n"]
else:
output = [f"Found {len(results)} result(s) for '{query}':\n"]
# Format search results
output = [f"Found {len(results)} result(s) for '{query}':\n"]
for i, result in enumerate(results[:20], 1): # Limit to 20 results
for i, result in enumerate(results, 1):
path = result.get("path", "unknown")
snippet = result.get("snippet", "")
matches = result.get("matches", [])
output.append(f"{i}. **{path}**")
if snippet:
output.append(f" {snippet[:200]}...")
# Show snippets with line numbers
for match in matches[:2]: # Limit to 2 snippets per note
line_num = match.get("line_number", "?")
context = match.get("context", "")
# Strip HTML tags for cleaner AI output
clean_context = re.sub(r'<[^>]+>', '', context)
if clean_context:
output.append(f" Line {line_num}: {clean_context[:150]}")
output.append("")
if len(results) > 20:
output.append(f"... and {len(results) - 20} more results")
# Add pagination hint if there are more results
if pagination and pagination.get("has_more"):
output.append(f"... more results available (use max_results and offset to paginate)")
return "\n".join(output)
def _tool_list_notes(self, args: dict) -> str:
"""List all notes."""
response = self.client.list_notes()
max_results = args.get("max_results")
offset = args.get("offset", 0)
# Use server-side pagination if max_results is specified
response = self.client.list_notes(limit=max_results, offset=offset)
if not response.success:
return f"Failed to list notes: {response.error}"
@ -246,8 +270,14 @@ class MCPServer:
data = response.data or {}
notes = data.get("notes", [])
folders = data.get("folders", [])
pagination = data.get("pagination")
output = [f"Found {len(notes)} note(s) in {len(folders)} folder(s):\n"]
# Build header with pagination info if available
if pagination:
total = pagination.get("total", len(notes))
output = [f"Found {total} note(s) in {len(folders)} folder(s) (showing {len(notes)}):\n"]
else:
output = [f"Found {len(notes)} note(s) in {len(folders)} folder(s):\n"]
# Group notes by folder
notes_by_folder: dict[str, list] = {}
@ -266,6 +296,10 @@ class MCPServer:
output.append(f" 📝 {name} (modified: {modified})")
output.append("")
# Add pagination hint if there are more results
if pagination and pagination.get("has_more"):
output.append(f"... more notes available (use max_results to limit)")
return "\n".join(output)
def _tool_get_note(self, args: dict) -> str:
@ -316,25 +350,40 @@ class MCPServer:
def _tool_get_notes_by_tag(self, args: dict) -> str:
"""Get notes with a specific tag."""
tag = args.get("tag", "")
max_results = args.get("max_results")
offset = args.get("offset", 0)
if not tag:
return "Error: tag is required"
response = self.client.get_notes_by_tag(tag)
# Use server-side pagination if max_results is specified
response = self.client.get_notes_by_tag(tag, limit=max_results, offset=offset)
if not response.success:
return f"Failed to get notes by tag: {response.error}"
data = response.data or {}
notes = data.get("notes", [])
count = data.get("count", len(notes))
pagination = data.get("pagination")
if not notes:
return f"No notes found with tag '#{tag}'"
# Build header with pagination info if available
if pagination:
output = [f"Notes with tag #{tag} ({count} total, showing {len(notes)}):\n"]
else:
output = [f"Notes with tag #{tag} ({count} total):\n"]
output = [f"Notes with tag #{tag}:\n"]
for note in notes:
path = note.get("path", "unknown")
output.append(f" 📝 {path}")
# Add pagination hint if there are more results
if pagination and pagination.get("has_more"):
output.append(f"\n... more notes available (use max_results to limit)")
return "\n".join(output)
def _tool_get_graph(self, args: dict) -> str:

27
mcp_server/tools.py
View File

@ -21,6 +21,14 @@ TOOLS: list[dict[str, Any]] = [
"query": {
"type": "string",
"description": "Search query. Can be keywords, phrases, or natural language."
},
"max_results": {
"type": "integer",
"description": "Maximum number of results to return. Useful for large vaults. If not specified, returns all matches."
},
"offset": {
"type": "integer",
"description": "Number of results to skip. Use with max_results for pagination."
}
},
"required": ["query"]
@ -31,7 +39,16 @@ TOOLS: list[dict[str, Any]] = [
"description": "List all notes in the knowledge base with their metadata (title, path, last modified date, size). Use this to get an overview of available notes or find notes by browsing.",
"inputSchema": {
"type": "object",
"properties": {},
"properties": {
"max_results": {
"type": "integer",
"description": "Maximum number of notes to return. Useful for large vaults. If not specified, returns all notes."
},
"offset": {
"type": "integer",
"description": "Number of notes to skip. Use with max_results for pagination."
}
},
"required": []
}
},
@ -71,6 +88,14 @@ TOOLS: list[dict[str, Any]] = [
"tag": {
"type": "string",
"description": "Tag name (without the # symbol)"
},
"max_results": {
"type": "integer",
"description": "Maximum number of notes to return. Useful for large vaults. If not specified, returns all matches."
},
"offset": {
"type": "integer",
"description": "Number of notes to skip. Use with max_results for pagination."
}
},
"required": ["tag"]