"""
Synchronous JMAP client.
Wraps session establishment, HTTP communication, and method dispatching
into a single object with a clean public API.
Auth note: JMAP has no 401-challenge-retry dance (unlike CalDAV).
Credentials are sent upfront on every request. A 401/403 is a hard failure.
"""
from __future__ import annotations
import logging
import uuid
try:
import niquests as requests
from niquests.auth import HTTPBasicAuth
except ImportError:
import requests # type: ignore[no-redef]
from requests.auth import HTTPBasicAuth # type: ignore[no-redef]
from caldav.jmap._methods.calendar import build_calendar_get, parse_calendar_get
from caldav.jmap._methods.event import (
build_event_changes,
build_event_get,
build_event_query,
build_event_set_create,
build_event_set_destroy,
build_event_set_update,
parse_event_changes,
parse_event_get,
parse_event_set,
)
from caldav.jmap._methods.task import (
build_task_get,
build_task_list_get,
build_task_set_create,
build_task_set_destroy,
build_task_set_update,
parse_task_list_get,
parse_task_set,
)
from caldav.jmap.constants import CALENDAR_CAPABILITY, CORE_CAPABILITY, TASK_CAPABILITY
from caldav.jmap.convert import ical_to_jscal
from caldav.jmap.error import JMAPAuthError, JMAPMethodError
from caldav.jmap.objects.calendar import JMAPCalendar
from caldav.jmap.objects.calendar_object import JMAPCalendarObject
from caldav.jmap.session import Session, fetch_session
from caldav.requests import HTTPBearerAuth
log = logging.getLogger("caldav.jmap")
_DEFAULT_USING = [CORE_CAPABILITY, CALENDAR_CAPABILITY]
_TASK_USING = [CORE_CAPABILITY, TASK_CAPABILITY]
class _JMAPClientBase:
def __init__(
self,
url: str,
username: str | None = None,
password: str | None = None,
auth=None,
auth_type: str | None = None,
timeout: int = 30,
) -> None:
self.url = url
self.username = username
self.password = password
self.timeout = timeout
self._session_cache: Session | None = None
if auth is not None:
self._auth = auth
else:
self._auth = self._build_auth(auth_type)
def _build_auth(self, auth_type: str | None):
"""Select and construct the auth object.
**The JMAP support is experimental, the API may change in minor-releases**
JMAP supports Basic and Bearer auth; Digest is not supported.
When ``auth_type`` is ``None`` the type is inferred from the
credentials supplied: a username triggers Basic, a password
alone triggers Bearer, and neither raises :class:`JMAPAuthError`.
"""
effective_type = auth_type
if effective_type is None:
if self.username:
effective_type = "basic"
elif self.password:
effective_type = "bearer"
else:
raise JMAPAuthError(
url=self.url,
reason="No credentials provided. Supply username+password or a bearer token.",
)
if effective_type == "basic":
if not self.username or not self.password:
raise JMAPAuthError(
url=self.url,
reason="Basic auth requires both username and password.",
)
return HTTPBasicAuth(self.username, self.password)
elif effective_type == "bearer":
if not self.password:
raise JMAPAuthError(
url=self.url,
reason="Bearer auth requires a token supplied as the password argument.",
)
return HTTPBearerAuth(self.password)
else:
raise JMAPAuthError(
url=self.url,
reason=f"Unsupported auth_type {effective_type!r}. Use 'basic' or 'bearer'.",
)
def _raise_set_error(self, session: Session, err: dict) -> None:
raise JMAPMethodError(
url=session.api_url,
reason=f"set failed: {err}",
error_type=err.get("type", "serverError"),
)
@staticmethod
def _build_event_search_calls(
account_id: str,
calendar_id: str | None,
start: str | None,
end: str | None,
text: str | None,
) -> list[tuple]:
"""Return a batched [CalendarEvent/query, CalendarEvent/get] call list for _search."""
filter_dict: dict = {}
if calendar_id is not None:
filter_dict["inCalendars"] = [calendar_id]
if start is not None:
filter_dict["after"] = start
if end is not None:
filter_dict["before"] = end
if text is not None:
filter_dict["text"] = text
query_call = build_event_query(account_id, filter_condition=filter_dict or None)
get_call = (
"CalendarEvent/get",
{
"accountId": account_id,
"#ids": {
"resultOf": "ev-query-0",
"name": "CalendarEvent/query",
"path": "/ids",
},
},
"ev-get-1",
)
return [query_call, get_call]
[docs]
class JMAPClient(_JMAPClientBase):
"""Synchronous JMAP client for calendar operations.
Usage::
from caldav.jmap import get_jmap_client
client = get_jmap_client(url="https://jmap.example.com/.well-known/jmap",
username="alice", password="secret")
calendars = client.get_calendars()
Args:
url: URL of the JMAP session endpoint (``/.well-known/jmap``).
username: Username for Basic auth.
password: Password for Basic auth, or bearer token if no username.
auth: A pre-built requests-compatible auth object. Takes precedence
over username/password if provided.
auth_type: Force a specific auth type: ``"basic"`` or ``"bearer"``.
timeout: HTTP request timeout in seconds.
"""
def __enter__(self) -> JMAPClient:
return self
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
return None
def _get_session(self) -> Session:
"""Return the cached Session, fetching it on first call."""
if self._session_cache is None:
self._session_cache = fetch_session(self.url, auth=self._auth, timeout=self.timeout)
return self._session_cache
def _request(self, method_calls: list[tuple], using: list[str] | None = None) -> list:
"""POST a batch of JMAP method calls and return the methodResponses.
Args:
method_calls: List of 3-tuples ``(method_name, args_dict, call_id)``.
using: Capability URN list for the ``using`` field. Defaults to
``_DEFAULT_USING`` (core + calendars).
Returns:
List of 3-tuples ``(method_name, response_args, call_id)`` from
the server's ``methodResponses`` array.
Raises:
JMAPAuthError: On HTTP 401 or 403.
JMAPMethodError: If any methodResponse is an ``error`` response.
requests.HTTPError: On other non-2xx HTTP responses.
"""
session = self._get_session()
payload = {
"using": using if using is not None else _DEFAULT_USING,
"methodCalls": list(method_calls),
}
log.debug("JMAP POST to %s: %d method call(s)", session.api_url, len(method_calls))
response = requests.post(
session.api_url,
json=payload,
auth=self._auth,
headers={"Content-Type": "application/json", "Accept": "application/json"},
timeout=self.timeout,
)
if response.status_code in (401, 403):
raise JMAPAuthError(
url=session.api_url,
reason=f"HTTP {response.status_code} from API endpoint",
)
response.raise_for_status()
data = response.json()
method_responses = data.get("methodResponses", [])
for resp in method_responses:
method_name, resp_args, call_id = resp
if method_name == "error":
error_type = resp_args.get("type", "serverError")
raise JMAPMethodError(
url=session.api_url,
reason=f"Method call failed: {resp_args}",
error_type=error_type,
)
return method_responses
[docs]
def get_calendars(self) -> list[JMAPCalendar]:
"""Fetch all calendars for the authenticated account.
Returns:
List of :class:`~caldav.jmap.objects.calendar.JMAPCalendar` objects.
"""
session = self._get_session()
call = build_calendar_get(session.account_id)
responses = self._request([call])
for method_name, resp_args, _ in responses:
if method_name == "Calendar/get":
calendars = parse_calendar_get(resp_args)
for cal in calendars:
cal._client = self
cal._is_async = False
return calendars
return []
[docs]
def create_event(self, calendar_id: str, ical_str: str) -> str:
"""Create a calendar event from an iCalendar string.
Args:
calendar_id: The JMAP calendar ID to create the event in.
ical_str: A VCALENDAR string representing the event.
Returns:
The server-assigned JMAP event ID.
Raises:
JMAPMethodError: If the server rejects the create request.
"""
session = self._get_session()
jscal = ical_to_jscal(ical_str, calendar_id=calendar_id)
call = build_event_set_create(session.account_id, {"new-0": jscal})
responses = self._request([call])
for method_name, resp_args, _ in responses:
if method_name == "CalendarEvent/set":
created, _, _, not_created, _, _ = parse_event_set(resp_args)
if "new-0" in not_created:
self._raise_set_error(session, not_created["new-0"])
if "new-0" not in created:
raise JMAPMethodError(
url=session.api_url,
reason="CalendarEvent/set response missing created entry for new-0",
)
return created["new-0"]["id"]
raise JMAPMethodError(url=session.api_url, reason="No CalendarEvent/set response")
[docs]
def get_event(self, event_id: str) -> JMAPCalendarObject:
"""Fetch a calendar event by JMAP event ID.
Args:
event_id: The JMAP event ID to retrieve.
Returns:
A :class:`~caldav.jmap.objects.calendar_object.JMAPCalendarObject`
wrapping the raw JSCalendar dict. ``parent`` is ``None`` since
no :class:`~caldav.jmap.objects.calendar.JMAPCalendar` is available
at the client level.
Raises:
JMAPMethodError: If the event is not found.
"""
session = self._get_session()
call = build_event_get(session.account_id, ids=[event_id])
responses = self._request([call])
for method_name, resp_args, _ in responses:
if method_name == "CalendarEvent/get":
items = parse_event_get(resp_args)
if not items:
raise JMAPMethodError(
url=session.api_url,
reason=f"Event not found: {event_id}",
error_type="notFound",
)
return JMAPCalendarObject(data=items[0], parent=None)
raise JMAPMethodError(url=session.api_url, reason="No CalendarEvent/get response")
[docs]
def update_event(self, event_id: str, ical_str: str) -> None:
"""Update a calendar event from an iCalendar string.
Args:
event_id: The JMAP event ID to update.
ical_str: A VCALENDAR string with the updated event data.
Raises:
JMAPMethodError: If the server rejects the update.
"""
session = self._get_session()
patch = ical_to_jscal(ical_str)
patch.pop("uid", None) # uid is server-immutable after creation; patch must omit it
call = build_event_set_update(session.account_id, {event_id: patch})
responses = self._request([call])
for method_name, resp_args, _ in responses:
if method_name == "CalendarEvent/set":
_, _, _, _, not_updated, _ = parse_event_set(resp_args)
if event_id in not_updated:
self._raise_set_error(session, not_updated[event_id])
return
raise JMAPMethodError(url=session.api_url, reason="No CalendarEvent/set response")
def _search(
self,
calendar_id: str | None = None,
start: str | None = None,
end: str | None = None,
text: str | None = None,
parent: JMAPCalendar | None = None,
) -> list[JMAPCalendarObject]:
session = self._get_session()
calls = self._build_event_search_calls(session.account_id, calendar_id, start, end, text)
responses = self._request(calls)
for method_name, resp_args, _ in responses:
if method_name == "CalendarEvent/get":
return [
JMAPCalendarObject(data=item, parent=parent)
for item in parse_event_get(resp_args)
]
return []
[docs]
def search_events(
self,
calendar_id: str | None = None,
start: str | None = None,
end: str | None = None,
text: str | None = None,
) -> list[JMAPCalendarObject]:
"""Search for calendar events.
All parameters are optional; omitting all returns every event in the account.
Results are fetched in a single batched JMAP request using a result reference
from ``CalendarEvent/query`` into ``CalendarEvent/get``.
Args:
calendar_id: Limit results to this calendar.
start: Only events ending after this datetime (``YYYY-MM-DDTHH:MM:SS``).
end: Only events starting before this datetime (``YYYY-MM-DDTHH:MM:SS``).
text: Free-text search across title, description, locations, and participants.
Returns:
List of :class:`~caldav.jmap.objects.calendar_object.JMAPCalendarObject`
instances. ``parent`` is ``None`` on these objects since no
:class:`~caldav.jmap.objects.calendar.JMAPCalendar` is available at
the client level; use :meth:`JMAPCalendar.search` if you need ``parent``
set.
"""
return self._search(calendar_id=calendar_id, start=start, end=end, text=text)
[docs]
def get_sync_token(self) -> str:
"""Return the current CalendarEvent state string for use as a sync token.
Calls ``CalendarEvent/get`` with an empty ID list — no event data is
transferred, only the ``state`` field from the response.
Returns:
Opaque state string. Pass to :meth:`get_objects_by_sync_token` to
retrieve only what changed since this point.
"""
session = self._get_session()
call = build_event_get(session.account_id, ids=[])
responses = self._request([call])
for method_name, resp_args, _ in responses:
if method_name == "CalendarEvent/get":
return resp_args.get("state", "")
raise JMAPMethodError(url=session.api_url, reason="No CalendarEvent/get response")
[docs]
def get_objects_by_sync_token(
self, sync_token: str
) -> tuple[list[JMAPCalendarObject], list[JMAPCalendarObject], list[str]]:
"""Fetch events changed since a previous sync token.
Calls ``CalendarEvent/changes`` to discover which events were created,
modified, or destroyed since ``sync_token`` was issued. Created and
modified events are returned as
:class:`~caldav.jmap.objects.calendar_object.JMAPCalendarObject` instances;
destroyed events are returned as IDs (the objects no longer exist on the server).
Args:
sync_token: A state string previously returned by :meth:`get_sync_token`
or by a prior call to this method.
Returns:
A 3-tuple ``(added, modified, deleted)``:
- ``added``: objects for newly created events (``parent`` is ``None``).
- ``modified``: objects for updated events (``parent`` is ``None``).
- ``deleted``: Event IDs that were destroyed.
Raises:
JMAPMethodError: If the server reports ``hasMoreChanges: true``.
"""
session = self._get_session()
changes_call = build_event_changes(session.account_id, sync_token)
responses = self._request([changes_call])
created_ids: list[str] = []
updated_ids: list[str] = []
destroyed: list[str] = []
for method_name, resp_args, _ in responses:
if method_name == "CalendarEvent/changes":
_, _, has_more, created_ids, updated_ids, destroyed = parse_event_changes(resp_args)
if has_more:
raise JMAPMethodError(
url=session.api_url,
reason=(
"CalendarEvent/changes response was truncated by the server "
"(hasMoreChanges=true). Call get_sync_token() to obtain a "
"fresh baseline and re-sync."
),
error_type="serverPartialFail",
)
fetch_ids = created_ids + updated_ids
if not fetch_ids:
return [], [], destroyed
get_call = build_event_get(session.account_id, ids=fetch_ids)
get_responses = self._request([get_call])
events_by_id: dict[str, JMAPCalendarObject] = {}
for method_name, resp_args, _ in get_responses:
if method_name == "CalendarEvent/get":
for item in parse_event_get(resp_args):
events_by_id[item["id"]] = JMAPCalendarObject(data=item, parent=None)
added = [events_by_id[i] for i in created_ids if i in events_by_id]
modified = [events_by_id[i] for i in updated_ids if i in events_by_id]
return added, modified, destroyed
[docs]
def delete_event(self, event_id: str) -> None:
"""Delete a calendar event.
Args:
event_id: The JMAP event ID to delete.
Raises:
JMAPMethodError: If the server rejects the delete.
"""
session = self._get_session()
call = build_event_set_destroy(session.account_id, [event_id])
responses = self._request([call])
for method_name, resp_args, _ in responses:
if method_name == "CalendarEvent/set":
_, _, _, _, _, not_destroyed = parse_event_set(resp_args)
if event_id in not_destroyed:
self._raise_set_error(session, not_destroyed[event_id])
return
raise JMAPMethodError(url=session.api_url, reason="No CalendarEvent/set response")
def _get_object_by_uid(
self, uid: str, calendar_id: str | None = None, parent: JMAPCalendar | None = None
) -> JMAPCalendarObject:
# RFC 8984 FilterCondition has no uid field; UID matching is done client-side.
for obj in self._search(calendar_id=calendar_id, parent=parent):
if obj.data.get("uid") == uid:
return obj
raise JMAPMethodError(
url=self._get_session().api_url, reason=f"No calendar object found with UID: {uid}"
)
[docs]
def get_task_lists(self) -> list[dict]:
"""Fetch all task lists for the authenticated account.
Returns:
List of raw JMAP TaskList dicts as returned by the server.
"""
session = self._get_session()
call = build_task_list_get(session.account_id)
responses = self._request([call], using=_TASK_USING)
for method_name, resp_args, _ in responses:
if method_name == "TaskList/get":
return parse_task_list_get(resp_args)
return []
[docs]
def create_task(self, task_list_id: str, title: str, **kwargs) -> str:
"""Create a task in a task list.
Args:
task_list_id: The JMAP task list ID to create the task in.
title: Task title (maps to VTODO ``SUMMARY``).
**kwargs: Optional JMAP Task fields using wire names: ``description``,
``due``, ``start``, ``timeZone``, ``estimatedDuration``,
``percentComplete``, ``progress``, ``priority``.
Returns:
The server-assigned JMAP task ID.
Raises:
JMAPMethodError: If the server rejects the create request.
"""
session = self._get_session()
task_dict = {
"@type": "Task",
"uid": str(uuid.uuid4()),
"taskListId": task_list_id,
"title": title,
"percentComplete": 0,
"progress": "needs-action",
"priority": 0,
}
task_dict.update(kwargs)
call = build_task_set_create(session.account_id, {"new-0": task_dict})
responses = self._request([call], using=_TASK_USING)
for method_name, resp_args, _ in responses:
if method_name == "Task/set":
created, _, _, not_created, _, _ = parse_task_set(resp_args)
if "new-0" in not_created:
self._raise_set_error(session, not_created["new-0"])
return created["new-0"]["id"]
raise JMAPMethodError(url=session.api_url, reason="No Task/set response")
[docs]
def get_task(self, task_id: str) -> dict:
"""Fetch a task by ID.
Args:
task_id: The JMAP task ID to retrieve.
Returns:
Raw JMAP Task dict as returned by the server.
Raises:
JMAPMethodError: If the task is not found.
"""
session = self._get_session()
call = build_task_get(session.account_id, ids=[task_id])
responses = self._request([call], using=_TASK_USING)
for method_name, resp_args, _ in responses:
if method_name == "Task/get":
items = resp_args.get("list", [])
if not items:
raise JMAPMethodError(
url=session.api_url,
reason=f"Task not found: {task_id}",
error_type="notFound",
)
return items[0]
raise JMAPMethodError(url=session.api_url, reason="No Task/get response")
[docs]
def update_task(self, task_id: str, patch: dict) -> None:
"""Update a task with a partial patch.
Args:
task_id: The JMAP task ID to update.
patch: Partial patch dict mapping property names to new values.
Raises:
JMAPMethodError: If the server rejects the update.
"""
session = self._get_session()
call = build_task_set_update(session.account_id, {task_id: patch})
responses = self._request([call], using=_TASK_USING)
for method_name, resp_args, _ in responses:
if method_name == "Task/set":
_, _, _, _, not_updated, _ = parse_task_set(resp_args)
if task_id in not_updated:
self._raise_set_error(session, not_updated[task_id])
return
raise JMAPMethodError(url=session.api_url, reason="No Task/set response")
[docs]
def delete_task(self, task_id: str) -> None:
"""Delete a task.
Args:
task_id: The JMAP task ID to delete.
Raises:
JMAPMethodError: If the server rejects the delete.
"""
session = self._get_session()
call = build_task_set_destroy(session.account_id, [task_id])
responses = self._request([call], using=_TASK_USING)
for method_name, resp_args, _ in responses:
if method_name == "Task/set":
_, _, _, _, _, not_destroyed = parse_task_set(resp_args)
if task_id in not_destroyed:
self._raise_set_error(session, not_destroyed[task_id])
return
raise JMAPMethodError(url=session.api_url, reason="No Task/set response")