open-webui/backend/open_webui/routers/notes.py

import json
import logging
from typing import Optional


from fastapi import APIRouter, Depends, HTTPException, Request, status, BackgroundTasks
from pydantic import BaseModel

from open_webui.models.users import Users, UserNameResponse
from open_webui.models.notes import Notes, NoteModel, NoteForm, NoteUserResponse

from open_webui.config import ENABLE_ADMIN_CHAT_ACCESS, ENABLE_ADMIN_EXPORT
from open_webui.constants import ERROR_MESSAGES
from open_webui.env import SRC_LOG_LEVELS


from open_webui.utils.auth import get_admin_user, get_verified_user
from open_webui.utils.access_control import has_access

log = logging.getLogger(__name__)
log.setLevel(SRC_LOG_LEVELS["MODELS"])

router = APIRouter()

############################
# GetNotes
############################


@router.get("/", response_model=list[NoteUserResponse])
async def get_notes(user=Depends(get_verified_user)):
    notes = [
        NoteUserResponse(
            **{
                **note.model_dump(),
                "user": UserNameResponse(
                    **Users.get_user_by_id(note.user_id).model_dump()
                ),
            }
        )
        for note in Notes.get_notes_by_user_id(user.id, "write")
    ]

    return notes


@router.get("/list", response_model=list[NoteUserResponse])
async def get_note_list(user=Depends(get_verified_user)):
    notes = [
        NoteUserResponse(
            **{
                **note.model_dump(),
                "user": UserNameResponse(
                    **Users.get_user_by_id(note.user_id).model_dump()
                ),
            }
        )
        for note in Notes.get_notes_by_user_id(user.id, "read")
    ]

    return notes


############################
# CreateNewNote
############################


@router.post("/create", response_model=Optional[NoteModel])
async def create_new_note(form_data: NoteForm, user=Depends(get_admin_user)):
    try:
        note = Notes.insert_new_note(form_data, user.id)
        return note
    except Exception as e:
        log.exception(e)
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT()
        )


############################
# GetNoteById
############################


@router.get("/{id}", response_model=Optional[NoteModel])
async def get_note_by_id(id: str, user=Depends(get_verified_user)):
    note = Notes.get_note_by_id(id)
    if not note:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND
        )

    if user.role != "admin" and not has_access(
        user.id, type="read", access_control=note.access_control
    ):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT()
        )

    return note


############################
# UpdateNoteById
############################


@router.post("/{id}/update", response_model=Optional[NoteModel])
async def update_note_by_id(
    id: str, form_data: NoteForm, user=Depends(get_verified_user)
):
    note = Notes.get_note_by_id(id)
    if not note:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND
        )

    if user.role != "admin" and not has_access(
        user.id, type="write", access_control=note.access_control
    ):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT()
        )

    try:
        note = Notes.update_note_by_id(id, form_data)
        return note
    except Exception as e:
        log.exception(e)
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT()
        )


############################
# DeleteNoteById
############################


@router.delete("/{id}/delete", response_model=bool)
async def delete_note_by_id(id: str, user=Depends(get_verified_user)):
    note = Notes.get_note_by_id(id)
    if not note:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND
        )

    if user.role != "admin" and not has_access(
        user.id, type="write", access_control=note.access_control
    ):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT()
        )

    try:
        note = Notes.delete_note_by_id(id)
        return True
    except Exception as e:
        log.exception(e)
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT()
        )