Source code for nc_py_api.notes

"""Notes App API wrapper."""

import dataclasses
import datetime
import json
import typing

import httpx

from ._exceptions import check_error
from ._misc import check_capabilities, clear_from_params_empty, require_capabilities
from ._session import AsyncNcSessionBasic, NcSessionBasic


[docs] @dataclasses.dataclass class Note: """Class representing one **Note**.""" def __init__(self, raw_data: dict): self._raw_data = raw_data @property def note_id(self) -> int: """Unique identifier which is created by the server.""" return self._raw_data["id"] @property def etag(self) -> str: """The note's entity tag (ETag) indicates if a note's attribute has changed.""" return self._raw_data.get("etag", "") @property def readonly(self) -> bool: """Indicates if the note is read-only.""" return self._raw_data.get("readonly", False) @property def title(self) -> str: """The note's title is also used as filename for the note's file. .. note :: Some special characters are automatically removed and a sequential number is added if a note with the same title in the same category exists. """ return self._raw_data.get("title", "") @property def content(self) -> str: """Notes can contain arbitrary text. .. note:: Formatting should be done using Markdown, but not every markup can be supported by every client. """ return self._raw_data.get("content", "") @property def category(self) -> str: """Every note is assigned to a category. By default, the category is an empty string (not null), which means the note is uncategorized. .. note:: Categories are mapped to folders in the file backend. Illegal characters are automatically removed and the respective folder is automatically created. Sub-categories (mapped to sub-folders) can be created by using ``/`` as delimiter. """ return self._raw_data.get("category", "") @property def favorite(self) -> bool: """If a note is marked as favorite, it is displayed at the top of the notes' list.""" return self._raw_data.get("favorite", False) @property def last_modified(self) -> datetime.datetime: """Last modified date/time of the note. If not provided on note creation or content update, the current time is used. """ modified = self._raw_data.get("modified", 0) return datetime.datetime.utcfromtimestamp(modified).replace(tzinfo=datetime.timezone.utc) def __repr__(self): return f"<{self.__class__.__name__} id={self.note_id}, title={self.title}, last_modified={self.last_modified}>"
[docs] class NotesSettings(typing.TypedDict): """Settings of Notes App.""" notes_path: str """Path to the folder, where note's files are stored in Nextcloud. The path must be relative to the user folder. Default is the localized string **Notes**.""" file_suffix: str """Newly created note's files will have this file suffix. Default is **.txt**."""
[docs] class _NotesAPI: """Class implementing Nextcloud Notes API.""" _ep_base: str = "/index.php/apps/notes/api/v1" # without `index.php` we will get 405 error. last_etag: str """Used by ``get_list``, when **etag** param is ``True``.""" def __init__(self, session: NcSessionBasic): self._session = session self.last_etag = "" @property def available(self) -> bool: """Returns True if the Nextcloud instance supports this feature, False otherwise.""" return not check_capabilities("notes", self._session.capabilities)
[docs] def get_list( self, category: str | None = None, modified_since: int | None = None, limit: int | None = None, cursor: str | None = None, no_content: bool = False, etag: bool = False, ) -> list[Note]: """Get information of all Notes. :param category: Filter the result by category name. Notes with another category are not included in the result. :param modified_since: When provided only results newer than given Unix timestamp are returned. :param limit: Limit response to contain no more than the given number of notes. If there are more notes, then the result is chunked and the HTTP response header **X-Notes-Chunk-Cursor** is sent with a string value. .. note:: Use :py:attr:`~nc_py_api.nextcloud.Nextcloud.response_headers` property to achieve that. :param cursor: You should use the string value from the last request's HTTP response header ``X-Notes-Chunk-Cursor`` in order to get the next chunk of notes. :param no_content: Flag indicating should ``content`` field be excluded from response. :param etag: Flag indicating should ``ETag`` from last call be used. Default = **False**. """ require_capabilities("notes", self._session.capabilities) params = { "category": category, "pruneBefore": modified_since, "exclude": "content" if no_content else None, "chunkSize": limit, "chunkCursor": cursor, } clear_from_params_empty(list(params.keys()), params) headers = {"If-None-Match": self.last_etag} if self.last_etag and etag else {} r = _res_to_json(self._session.adapter.get(self._ep_base + "/notes", params=params, headers=headers)) self.last_etag = self._session.response_headers["ETag"] return [Note(i) for i in r]
[docs] def by_id(self, note: Note) -> Note: """Get updated information about :py:class:`~nc_py_api.notes.Note`.""" require_capabilities("notes", self._session.capabilities) r = _res_to_json( self._session.adapter.get( self._ep_base + f"/notes/{note.note_id}", headers={"If-None-Match": f'"{note.etag}"'} ) ) return Note(r) if r else note
[docs] def create( self, title: str, content: str | None = None, category: str | None = None, favorite: bool | None = None, last_modified: int | str | datetime.datetime | None = None, ) -> Note: """Create new Note.""" require_capabilities("notes", self._session.capabilities) params = { "title": title, "content": content, "category": category, "favorite": favorite, "modified": last_modified, } clear_from_params_empty(list(params.keys()), params) return Note(_res_to_json(self._session.adapter.post(self._ep_base + "/notes", json=params)))
[docs] def update( self, note: Note, title: str | None = None, content: str | None = None, category: str | None = None, favorite: bool | None = None, overwrite: bool = False, ) -> Note: """Updates Note. ``overwrite`` specifies should be or not the Note updated even if it was changed on server(has different ETag). """ require_capabilities("notes", self._session.capabilities) headers = {"If-Match": f'"{note.etag}"'} if not overwrite else {} params = { "title": title, "content": content, "category": category, "favorite": favorite, } clear_from_params_empty(list(params.keys()), params) if not params: raise ValueError("Nothing to update.") return Note( _res_to_json( self._session.adapter.put(self._ep_base + f"/notes/{note.note_id}", json=params, headers=headers) ) )
[docs] def delete(self, note: int | Note) -> None: """Deletes a Note.""" require_capabilities("notes", self._session.capabilities) note_id = note.note_id if isinstance(note, Note) else note check_error(self._session.adapter.delete(self._ep_base + f"/notes/{note_id}"))
[docs] def get_settings(self) -> NotesSettings: """Returns Notes App settings.""" require_capabilities("notes", self._session.capabilities) r = _res_to_json(self._session.adapter.get(self._ep_base + "/settings")) return {"notes_path": r["notesPath"], "file_suffix": r["fileSuffix"]}
[docs] def set_settings(self, notes_path: str | None = None, file_suffix: str | None = None) -> None: """Change specified setting(s).""" if notes_path is None and file_suffix is None: raise ValueError("No setting to change.") require_capabilities("notes", self._session.capabilities) params = { "notesPath": notes_path, "fileSuffix": file_suffix, } clear_from_params_empty(list(params.keys()), params) check_error(self._session.adapter.put(self._ep_base + "/settings", json=params))
class _AsyncNotesAPI: """Class implements Async Nextcloud Notes API.""" _ep_base: str = "/index.php/apps/notes/api/v1" # without `index.php` we will get 405 error. last_etag: str """Used by ``get_list``, when **etag** param is ``True``.""" def __init__(self, session: AsyncNcSessionBasic): self._session = session self.last_etag = "" @property async def available(self) -> bool: """Returns True if the Nextcloud instance supports this feature, False otherwise.""" return not check_capabilities("notes", await self._session.capabilities) async def get_list( self, category: str | None = None, modified_since: int | None = None, limit: int | None = None, cursor: str | None = None, no_content: bool = False, etag: bool = False, ) -> list[Note]: """Get information of all Notes. :param category: Filter the result by category name. Notes with another category are not included in the result. :param modified_since: When provided only results newer than given Unix timestamp are returned. :param limit: Limit response to contain no more than the given number of notes. If there are more notes, then the result is chunked and the HTTP response header **X-Notes-Chunk-Cursor** is sent with a string value. .. note:: Use :py:attr:`~nc_py_api.nextcloud.Nextcloud.response_headers` property to achieve that. :param cursor: You should use the string value from the last request's HTTP response header ``X-Notes-Chunk-Cursor`` in order to get the next chunk of notes. :param no_content: Flag indicating should ``content`` field be excluded from response. :param etag: Flag indicating should ``ETag`` from last call be used. Default = **False**. """ require_capabilities("notes", await self._session.capabilities) params = { "category": category, "pruneBefore": modified_since, "exclude": "content" if no_content else None, "chunkSize": limit, "chunkCursor": cursor, } clear_from_params_empty(list(params.keys()), params) headers = {"If-None-Match": self.last_etag} if self.last_etag and etag else {} r = _res_to_json(await self._session.adapter.get(self._ep_base + "/notes", params=params, headers=headers)) self.last_etag = self._session.response_headers["ETag"] return [Note(i) for i in r] async def by_id(self, note: Note) -> Note: """Get updated information about :py:class:`~nc_py_api.notes.Note`.""" require_capabilities("notes", await self._session.capabilities) r = _res_to_json( await self._session.adapter.get( self._ep_base + f"/notes/{note.note_id}", headers={"If-None-Match": f'"{note.etag}"'} ) ) return Note(r) if r else note async def create( self, title: str, content: str | None = None, category: str | None = None, favorite: bool | None = None, last_modified: int | str | datetime.datetime | None = None, ) -> Note: """Create new Note.""" require_capabilities("notes", await self._session.capabilities) params = { "title": title, "content": content, "category": category, "favorite": favorite, "modified": last_modified, } clear_from_params_empty(list(params.keys()), params) return Note(_res_to_json(await self._session.adapter.post(self._ep_base + "/notes", json=params))) async def update( self, note: Note, title: str | None = None, content: str | None = None, category: str | None = None, favorite: bool | None = None, overwrite: bool = False, ) -> Note: """Updates Note. ``overwrite`` specifies should be or not the Note updated even if it was changed on server(has different ETag). """ require_capabilities("notes", await self._session.capabilities) headers = {"If-Match": f'"{note.etag}"'} if not overwrite else {} params = { "title": title, "content": content, "category": category, "favorite": favorite, } clear_from_params_empty(list(params.keys()), params) if not params: raise ValueError("Nothing to update.") return Note( _res_to_json( await self._session.adapter.put(self._ep_base + f"/notes/{note.note_id}", json=params, headers=headers) ) ) async def delete(self, note: int | Note) -> None: """Deletes a Note.""" require_capabilities("notes", await self._session.capabilities) note_id = note.note_id if isinstance(note, Note) else note check_error(await self._session.adapter.delete(self._ep_base + f"/notes/{note_id}")) async def get_settings(self) -> NotesSettings: """Returns Notes App settings.""" require_capabilities("notes", await self._session.capabilities) r = _res_to_json(await self._session.adapter.get(self._ep_base + "/settings")) return {"notes_path": r["notesPath"], "file_suffix": r["fileSuffix"]} async def set_settings(self, notes_path: str | None = None, file_suffix: str | None = None) -> None: """Change specified setting(s).""" if notes_path is None and file_suffix is None: raise ValueError("No setting to change.") require_capabilities("notes", await self._session.capabilities) params = { "notesPath": notes_path, "fileSuffix": file_suffix, } clear_from_params_empty(list(params.keys()), params) check_error(await self._session.adapter.put(self._ep_base + "/settings", json=params)) def _res_to_json(response: httpx.Response) -> dict: check_error(response) return json.loads(response.text) if response.status_code != 304 else {}