Spaces:
Sleeping
Sleeping
File size: 6,524 Bytes
f871fed |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 |
from typing import List, Literal, Optional
from fastapi import APIRouter, HTTPException, Query
from loguru import logger
from api.models import NoteCreate, NoteResponse, NoteUpdate
from open_notebook.domain.notebook import Note
from open_notebook.exceptions import InvalidInputError
router = APIRouter()
@router.get("/notes", response_model=List[NoteResponse])
async def get_notes(
notebook_id: Optional[str] = Query(None, description="Filter by notebook ID")
):
"""Get all notes with optional notebook filtering."""
try:
if notebook_id:
# Get notes for a specific notebook
from open_notebook.domain.notebook import Notebook
notebook = await Notebook.get(notebook_id)
if not notebook:
raise HTTPException(status_code=404, detail="Notebook not found")
notes = await notebook.get_notes()
else:
# Get all notes
notes = await Note.get_all(order_by="updated desc")
return [
NoteResponse(
id=note.id or "",
title=note.title,
content=note.content,
note_type=note.note_type,
created=str(note.created),
updated=str(note.updated),
)
for note in notes
]
except HTTPException:
raise
except Exception as e:
logger.error(f"Error fetching notes: {str(e)}")
raise HTTPException(status_code=500, detail=f"Error fetching notes: {str(e)}")
@router.post("/notes", response_model=NoteResponse)
async def create_note(note_data: NoteCreate):
"""Create a new note."""
try:
# Auto-generate title if not provided and it's an AI note
title = note_data.title
if not title and note_data.note_type == "ai" and note_data.content:
from open_notebook.graphs.prompt import graph as prompt_graph
prompt = "Based on the Note below, please provide a Title for this content, with max 15 words"
result = await prompt_graph.ainvoke(
{ # type: ignore[arg-type]
"input_text": note_data.content,
"prompt": prompt
}
)
title = result.get("output", "Untitled Note")
# Validate note_type
note_type: Optional[Literal["human", "ai"]] = None
if note_data.note_type in ("human", "ai"):
note_type = note_data.note_type # type: ignore[assignment]
elif note_data.note_type is not None:
raise HTTPException(status_code=400, detail="note_type must be 'human' or 'ai'")
new_note = Note(
title=title,
content=note_data.content,
note_type=note_type,
)
await new_note.save()
# Add to notebook if specified
if note_data.notebook_id:
from open_notebook.domain.notebook import Notebook
notebook = await Notebook.get(note_data.notebook_id)
if not notebook:
raise HTTPException(status_code=404, detail="Notebook not found")
await new_note.add_to_notebook(note_data.notebook_id)
return NoteResponse(
id=new_note.id or "",
title=new_note.title,
content=new_note.content,
note_type=new_note.note_type,
created=str(new_note.created),
updated=str(new_note.updated),
)
except HTTPException:
raise
except InvalidInputError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
logger.error(f"Error creating note: {str(e)}")
raise HTTPException(status_code=500, detail=f"Error creating note: {str(e)}")
@router.get("/notes/{note_id}", response_model=NoteResponse)
async def get_note(note_id: str):
"""Get a specific note by ID."""
try:
note = await Note.get(note_id)
if not note:
raise HTTPException(status_code=404, detail="Note not found")
return NoteResponse(
id=note.id or "",
title=note.title,
content=note.content,
note_type=note.note_type,
created=str(note.created),
updated=str(note.updated),
)
except HTTPException:
raise
except Exception as e:
logger.error(f"Error fetching note {note_id}: {str(e)}")
raise HTTPException(status_code=500, detail=f"Error fetching note: {str(e)}")
@router.put("/notes/{note_id}", response_model=NoteResponse)
async def update_note(note_id: str, note_update: NoteUpdate):
"""Update a note."""
try:
note = await Note.get(note_id)
if not note:
raise HTTPException(status_code=404, detail="Note not found")
# Update only provided fields
if note_update.title is not None:
note.title = note_update.title
if note_update.content is not None:
note.content = note_update.content
if note_update.note_type is not None:
if note_update.note_type in ("human", "ai"):
note.note_type = note_update.note_type # type: ignore[assignment]
else:
raise HTTPException(status_code=400, detail="note_type must be 'human' or 'ai'")
await note.save()
return NoteResponse(
id=note.id or "",
title=note.title,
content=note.content,
note_type=note.note_type,
created=str(note.created),
updated=str(note.updated),
)
except HTTPException:
raise
except InvalidInputError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
logger.error(f"Error updating note {note_id}: {str(e)}")
raise HTTPException(status_code=500, detail=f"Error updating note: {str(e)}")
@router.delete("/notes/{note_id}")
async def delete_note(note_id: str):
"""Delete a note."""
try:
note = await Note.get(note_id)
if not note:
raise HTTPException(status_code=404, detail="Note not found")
await note.delete()
return {"message": "Note deleted successfully"}
except HTTPException:
raise
except Exception as e:
logger.error(f"Error deleting note {note_id}: {str(e)}")
raise HTTPException(status_code=500, detail=f"Error deleting note: {str(e)}") |