from pathlib import PurePath
from typing import Optional, AsyncIterator, AsyncContextManager
from urllib.parse import quote as url_quote
from uuid import UUID, uuid4
from .error import ApiError
from .parser import DavParser
from .request import Requester, Request
from .task import TaskFile, TaskList
class NextcloudTasksApi(AsyncContextManager):
"""
Main API interface, using a backend for XML DAV requests, giving raw ICAL data.
Must be entered/exited (or opened/closed) to set up the request backend with authentication and connection pool.
Note that some endpoints do not return the updated or created content. To avoid the additional fetch roundtrip,
a local copy will be returned if possible (i.e., when the new ETag is available).
"""
def __init__(self, requester: Requester, parser: DavParser = DavParser()) -> None:
self._requester: Requester = requester
self._parser: DavParser = parser
async def list_user_principal(self) -> AsyncIterator[str]:
"""Query current user principal endpoints. Useful to validate endpoint and authentication settings."""
request: Request = Request(
method="PROPFIND",
url="remote.php/dav/",
headers={"Content-Type": "application/xml; charset=utf-8", "Depth": "0"},
content=self._parser.get_user_principal(),
)
async with self._requester.request(request) as response:
response.raise_for(207, "application/xml", "utf-8")
async for href in self._parser.parse_for_user_principal(response):
yield self._make_href(href).as_posix()
async def get_lists(self) -> AsyncIterator[TaskList]:
"""Find all available task lists."""
request: Request = Request(
method="PROPFIND",
url="remote.php/dav/calendars/{}/".format(self._url_quote(self._requester.username)),
headers={"Content-Type": "application/xml; charset=utf-8"},
content=self._parser.get_propfind_calendar_list(),
)
async with self._requester.request(request) as response:
response.raise_for(207, "application/xml", "utf-8")
async for href, name, color in self._parser.parse_for_calendar_list(response):
yield TaskList(href=self._make_href(href), name=name, color=color)
async def get_list(self, task_list: TaskList, completed: Optional[bool] = None) -> AsyncIterator[TaskFile]:
"""Get all tasks of a task list, optionally filter remotely by completion state."""
request: Request = Request(
method="REPORT",
url=task_list.href.as_posix(),
headers={"Content-Type": "application/xml; charset=utf-8", "Depth": "1"},
content=self._parser.get_report_calendars(completed),
)
async with self._requester.request(request) as response:
response.raise_for(207, "application/xml", "utf-8")
async for href, etag, caldav in self._parser.parse_for_calendars(response):
yield TaskFile(href=self._make_href(href), etag=etag, content=caldav)
async def create_list(self, name: str, color: str = "#0082c9", filename: Optional[str] = None) -> TaskList:
"""Create a new task list with the given displayname and (unverified) filename."""
url: PurePath = PurePath("remote.php/dav/calendars") / \
self._url_quote(self._requester.username) / \
self._url_quote(filename if filename is not None else name.lower())
request: Request = Request(
method="MKCOL",
url=url.as_posix(),
headers={"Content-Type": "application/xml; charset=utf-8", "Depth": "0"},
content=self._parser.get_mkcol_calendar(name, color),
)
async with self._requester.request(request) as response:
response.raise_for(201)
return TaskList(href=url, name=name, color=color)
async def update_list(self, task_list: TaskList) -> None:
"""Update a task list with its updated handle."""
request: Request = Request(
method="PROPPATCH",
url=task_list.href.as_posix(),
headers={"Content-Type": "application/xml; charset=utf-8", "Depth": "0"},
content=self._parser.get_property_update(
task_list.name if task_list.name is not None else task_list.href.stem,
task_list.color if task_list.color is not None else "#0082c9",
)
)
async with self._requester.request(request) as response:
response.raise_for(207, "application/xml", "utf-8")
if [self._make_href(_) async for _ in self._parser.parse_propstat_for_status(response)] != [task_list.href]:
raise ApiError(f"Cannot update tasklist '{task_list.href}'") from None
async def delete_list(self, task_list: TaskList) -> None:
"""Delete the given task list."""
request: Request = Request(
method="DELETE",
headers={"Depth": "0"},
url=task_list.href.as_posix(),
)
async with self._requester.request(request) as response:
response.raise_for(204)
async def create_task(self, task_list: TaskList, task: str, filename: Optional[UUID] = None) -> TaskFile:
"""Create a new task in the task list with the given iCal content."""
href: PurePath = task_list.href
href /= PurePath(str(filename if filename is not None else uuid4()).upper()).with_suffix(".ics")
request: Request = Request(
method="PUT",
url=href.as_posix(),
headers={"Content-Type": "text/calendar; charset=utf-8"},
content=task.encode(encoding="utf-8", errors="strict"),
)
async with self._requester.request(request) as response:
response.raise_for(201)
etag: Optional[str] = response.headers.get("oc-etag", response.headers.get("etag"))
if etag is not None:
return TaskFile(href=href, etag=etag, content=task)
return await self._get_task(href)
async def update_task(self, task: TaskFile) -> TaskFile:
"""Update a task with new content."""
request: Request = Request(
method="PUT",
url=task.href.as_posix(),
headers={**{"Content-Type": "text/calendar; component=vtodo; charset=utf-8"},
**({"If-Match": task.etag} if task.etag is not None else {})},
content=task.content.encode(encoding="utf-8", errors="strict"),
)
async with self._requester.request(request) as response:
response.raise_for(204)
etag: Optional[str] = response.headers.get("oc-etag", response.headers.get("etag"))
if etag is not None and task.etag is not None:
return TaskFile(href=task.href, etag=etag, content=task.content)
return await self._get_task(task.href)
async def delete_task(self, task: TaskFile) -> None:
"""Delete a task. This does not recursively delete subtasks."""
request: Request = Request(
method="DELETE",
headers={"If-Match": task.etag} if task.etag is not None else {},
url=task.href.as_posix(),
)
async with self._requester.request(request) as response:
response.raise_for(204)
async def _get_task(self, task_href: PurePath) -> TaskFile:
request: Request = Request(
method="PROPFIND",
url=task_href.as_posix(),
headers={"Content-Type": "application/xml; charset=utf-8"},
content=self._parser.get_propfind_calendar(),
)
async with self._requester.request(request) as response:
response.raise_for(207, "application/xml", "utf-8")
task = await self._parser.parse_for_calendar(response)
if task is None:
raise ApiError(f"Cannot get task '{task_href}'")
href, etag, caldav = task
return TaskFile(href=self._make_href(href), etag=etag, content=caldav)
@classmethod
def _make_href(cls, href: str) -> PurePath:
"""Treat all 'remote.php' API URLs as relative to configured base URL, even when returned absolute."""
return PurePath(href.lstrip("/") if href.startswith("/remote.php/") else href)
@classmethod
def _url_quote(cls, s: str) -> str:
"""Make the given string safe to be used in URL path context."""
return url_quote(s, safe='')
async def __aenter__(self) -> 'NextcloudTasksApi':
await self._requester.__aenter__()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
await self._requester.__aexit__(exc_type, exc_val, exc_tb)
async def open(self) -> None:
await self.__aenter__()
async def close(self) -> None:
await self.__aexit__(None, None, None)
@property
def username(self) -> str:
return self._requester.username
@property
def base_url(self) -> str:
return self._requester.base_url
def __str__(self) -> str:
return f"<{self.__class__.__name__} {self.username} @ {self.base_url}>"
