CalDAV Calendar Sync Client

CalDAV Calendar Sync Client

Problem Statement

Teams and individuals often maintain calendars across multiple CalDAV servers (Nextcloud, Radicale, Fastmail, Google via bridges, iCloud via app-specific passwords). Copying events between them by hand is error-prone, and there's no simple command-line way to pull a calendar down, inspect it, or feed it into automation (backups, alerting, dashboards, migration scripts). Existing GUI clients are heavyweight and don't fit into cron jobs or CI pipelines.

This script solves that by acting as a focused, scriptable CalDAV client: it connects to a server, discovers the requested calendar, fetches all events (including recurrences) in a configurable time window, and serializes them to JSON or iCalendar. It's designed to be the building block for backups, cross-server sync jobs, and downstream data processing.

How It Works

The script uses the caldav library, which wraps the CalDAV protocol (RFC 4791) on top of requests. The flow is: connect → discover principal → locate calendar → search events → serialize → write output. Connection uses HTTP Basic auth by default, but the caldav client transparently handles the .well-known/caldav redirect that most servers use to advertise their principal URL.

Design decisions:

• Search, not enumeration. We use calendar.search() with a start/end window rather than iterating every resource. This is dramatically faster on large calendars and lets us control scope.
• Expand recurrences. Recurring events are expanded server-side or client-side (depending on server support) so the JSON output contains concrete instances — useful for downstream consumers that don't understand RRULE.
• Two output formats. JSON for programmatic use, .ics for re-importing or archiving. Chosen via file extension or explicit flag.
• Env-var configuration. Credentials never go on the command line (which leaks into ps and shell history); they come from environment variables with sane defaults.

Full Script

#!/usr/bin/env python3
"""
CalDAV Calendar Sync Client

Connects to a CalDAV server, fetches events from a named calendar within a
configurable time window (recurrences expanded), and writes them to JSON or
iCalendar format. Suitable for cron jobs, backups, and pipeline integration.

Usage:
    python3 caldav_sync.py --url https://cal.example.com \
        --user admin --calendar Work --out events.json
"""

from __future__ import annotations

import argparse
import json
import logging
import os
import sys
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from typing import Any

import caldav
from caldav.elements import dav
from icalendar import Calendar as ICalCalendar

# ---------------------------------------------------------------------------
# Configuration (overridable via environment variables)
# ---------------------------------------------------------------------------
DEFAULT_URL = os.environ.get("CALDAV_URL", "")
DEFAULT_USER = os.environ.get("CALDAV_USER", "")
DEFAULT_PASSWORD = os.environ.get("CALDAV_PASSWORD", "")
DEFAULT_CALENDAR = os.environ.get("CALDAV_CALENDAR", "default")
DEFAULT_DAYS_BACK = int(os.environ.get("CALDAV_DAYS_BACK", "30"))
DEFAULT_DAYS_FORWARD = int(os.environ.get("CALDAV_DAYS_FORWARD", "365"))
DEFAULT_TIMEOUT = int(os.environ.get("CALDAV_TIMEOUT", "30"))

log = logging.getLogger("caldav_sync")


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    """Parse CLI arguments. Env vars provide defaults so secrets stay out of argv."""
    p = argparse.ArgumentParser(
        description="Fetch events from a CalDAV calendar and export them.",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    p.add_argument("--url", default=DEFAULT_URL, help="CalDAV server base URL")
    p.add_argument("--user", default=DEFAULT_USER, help="Username")
    p.add_argument("--password", default=DEFAULT_PASSWORD,
                   help="Password (prefer CALDAV_PASSWORD env var)")
    p.add_argument("--calendar", default=DEFAULT_CALENDAR,
                   help="Calendar name or ID to fetch")
    p.add_argument("--out", type=Path, required=True,
                   help="Output file (.json or .ics)")
    p.add_argument("--days-back", type=int, default=DEFAULT_DAYS_BACK,
                   help="How many days into the past to fetch")
    p.add_argument("--days-forward", type=int, default=DEFAULT_DAYS_FORWARD,
                   help="How many days into the future to fetch")
    p.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT,
                   help="Network timeout in seconds")
    p.add_argument("--verbose", "-v", action="store_true",
                   help="Enable debug logging")
    return p.parse_args(argv)


def connect(url: str, user: str, password: str, timeout: int) -> caldav.DAVClient:
    """Create an authenticated DAVClient, validating inputs early."""
    if not url:
        raise ValueError("CalDAV URL is required (--url or CALDAV_URL)")
    if not user or not password:
        raise ValueError("Credentials required (--user/--password or env vars)")
    # caldav uses `requests` underneath; timeout is passed per-request via the
    # client's session, so we set it on the DAVClient itself.
    return caldav.DAVClient(
        url=url,
        username=user,
        password=password,
        timeout=timeout,
    )


def find_calendar(principal: caldav.Principal, name: str) -> caldav.Calendar:
    """Return the first calendar whose name or URL basename matches `name`."""
    for cal in principal.calendars():
        # cal.get_display_name() returns the human-readable name; the URL
        # basename is a fallback for servers that don't set a display name.
        display = (cal.get_display_name() or "").strip()
        url_name = str(cal.url).rstrip("/").rsplit("/", 1)[-1]
        if display.lower() == name.lower() or url_name.lower() == name.lower():
            return cal
    available = [c.get_display_name() for c in principal.calendars()]
    raise LookupError(f"Calendar '{name}' not found. Available: {available}")


def event_to_dict(component: Any) -> dict[str, Any]:
    """Flatten an icalendar VTODO/VEVENT into a JSON-serializable dict."""
    def _str(key: str) -> str | None:
        val = component.get(key)
        return str(val) if val is not None else None

    def _dt(key: str) -> str | None:
        val = component.get(key)
        if val is None:
            return None
        # icalendar returns vDDDTypes; .dt gives datetime/date. Normalize to ISO.
        dt = getattr(val, "dt", val)
        if isinstance(dt, (datetime, date)):
            return dt.isoformat()
        return str(dt)

    return {
        "uid": _str("UID"),
        "summary": _str("SUMMARY"),
        "description": _str("DESCRIPTION"),
        "location": _str("LOCATION"),
        "status": _str("STATUS"),
        "start": _dt("DTSTART"),
        "end": _dt("DTEND") or _dt("DUE"),
        "created": _dt("CREATED"),
        "last_modified": _dt("LAST-MODIFIED"),
        "organizer": _str("ORGANIZER"),
        "categories": [str(c) for c in component.get("CATEGORIES", [])] or None,
        "raw": component.to_ical().decode("utf-8", errors="replace"),
    }


def fetch_events(calendar: caldav.Calendar, start: datetime, end: datetime) -> list[dict]:
    """Search for events in the window and expand recurrences into instances."""
    # `expand=True` asks the server to expand RRULEs into concrete instances.
    # Servers that don't support expansion will return the master event; we
    # accept that trade-off rather than reimplementing RRULE logic locally.
    results = calendar.search(
        start=start,
        end=end,
        event=True,
        expand=True,
    )

    events: list[dict] = []
    for resource in results:
        # Each resource may contain multiple components (e.g. VTIMEZONE + VEVENT).
        ical = ICalCalendar.from_ical(resource.data)
        for component in ical.walk():
            if component.name in ("VEVENT", "VTODO"):
                events.append(event_to_dict(component))

    # Stable ordering makes diffs between runs meaningful.
    events.sort(key=lambda e: (e.get("start") or "", e.get("uid") or ""))
    return events


def write_output(path: Path, events: list[dict], raw_ics: str | None) -> None:
    """Write JSON or ICS depending on the output file's extension."""
    path.parent.mkdir(parents=True, exist_ok=True)
    suffix = path.suffix.lower()

    if suffix == ".ics":
        if raw_ics is None:
            raise ValueError("Cannot write .ics output: no raw calendar data available")
        path.write_text(raw_ics, encoding="utf-8")
    elif suffix == ".json":
        payload = {
            "generated_at": datetime.now(timezone.utc).isoformat(),
            "count": len(events),
            "events": events,
        }
        path.write_text(json.dumps(payload, indent=2, default=str), encoding="utf-8")
    else:
        raise ValueError(f"Unsupported output extension: {suffix!r} (use .json or .ics)")


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------
def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    logging.basicConfig(
        level=logging.DEBUG if args.verbose else logging.INFO,
        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
    )

    now = datetime.now(timezone.utc)
    start = now - timedelta(days=args.days_back)
    end = now + timedelta(days=args.days_forward)

    try:
        log.info("Connecting to %s", args.url)
        client = connect(args.url, args.user, args.password, args.timeout)

        principal = client.principal()
        log.info("Authenticated as %s", principal.url)

        calendar = find_calendar(principal, args.calendar)
        log.info("Using calendar: %s", calendar.get_display_name() or calendar.url)

        log.info("Fetching events from %s to %s", start.date(), end.date())
        events = fetch_events(calendar, start, end)
        log.info("Fetched %d event(s)", len(events))

        # For .ics output we also collect the raw iCalendar text to preserve
        # fields the JSON flattener drops (VALARM, ATTENDEE, RRULE, etc.).
        raw_ics: str | None = None
        if args.out.suffix.lower() == ".ics":
            merged = ICalCalendar()
            merged.add("prodid", "-//caldav_sync//EN")
            merged.add("version", "2.0")
            for ev in events:
                try:
                    merged.add_component(ICalCalendar.from_ical(ev["raw"]))
                except Exception as exc:  # noqa: BLE001 — one bad event shouldn't kill the run
                    log.warning("Skipping unparseable event %s: %s", ev.get("uid"), exc)
            raw_ics = merged.to_ical().decode("utf-8")

        write_output(args.out, events, raw_ics)
        log.info("Wrote %s", args.out)
        return 0

    except caldav.lib.error.AuthorizationError as exc:
        log.error("Authentication failed: %s", exc)
        return 2
    except caldav.lib.error.NotFoundError as exc:
        log.error("CalDAV endpoint not found (check --url): %s", exc)
        return 3
    except (ValueError, LookupError) as exc:
        log.error("Configuration error: %s", exc)
        return 4
    except Exception as exc:  # noqa: BLE001 — top-level guard for clean exit codes
        log.exception("Unexpected failure: %s", exc)
        return 1


if __name__ == "__main__":
    sys.exit(main())

Configuration

All configuration can be provided via CLI flags or environment variables. Flags take precedence; env vars provide defaults so secrets stay out of shell history and ps output.

Output format is inferred from --out's extension: .json produces a structured payload, .ics produces a merged iCalendar file.

Usage Examples

1. Export the next year of a Work calendar to JSON:

export CALDAV_URL=https://cal.example.com
export CALDAV_USER=admin
export CALDAV_PASSWORD=secret
python3 caldav_sync.py --calendar Work --out events.json

2024-06-01 12:00:01 INFO caldav_sync: Connecting to https://cal.example.com
2024-06-01 12:00:02 INFO caldav_sync: Authenticated as /dav/principals/admin/
2024-06-01 12:00:02 INFO caldav_sync: Using calendar: Work
2024-06-01 12:00:02 INFO caldav_sync: Fetching events from 2024-05-02 to 2025-06-01
2024-06-01 12:00:04 INFO caldav_sync: Fetched 147 event(s)
2024-06-01 12:00:04 INFO caldav_sync: Wrote events.json

2. Nightly backup of a full calendar to .ics (cron-friendly):

CALDAV_PASSWORD="$TOKEN" python3 caldav_sync.py \
    --url https://radicale.internal/dav \
    --user svc-backup \
    --calendar Personal \
    --days-back 3650 --days-forward 3650 \
    --out /backups/personal-$(date +%F).ics

3. Narrow window for an on-call dashboard:

python3 caldav_sync.py --calendar OnCall --days-back 1 --days-forward 14 \
    --out /var/www/oncall.json -v

Error Handling

The script uses distinct exit codes so callers (cron, CI) can branch on failure type:

• 2 — AuthorizationError. Bad credentials. The message is logged without echoing the password.
• 3 — NotFoundError. The URL doesn't point at a CalDAV endpoint. Common when the base URL is wrong or the server doesn't expose .well-known/caldav.
• 4 — Configuration errors. Missing URL/credentials, unknown calendar name (the error lists available calendars), or unsupported output extension.
• 1 — Unexpected. Network timeouts, TLS errors, malformed server responses — logged with a full traceback for debugging.

Inside fetch_events, individual malformed events that fail to parse are logged as warnings and skipped rather than aborting the entire run — one bad resource shouldn't invalidate a backup. Server-side errors during search bubble up and are caught by the top-level handler. All network calls honor CALDAV_TIMEOUT, preventing hung cron jobs.

Extending Ideas

• Two-way sync. Diff the fetched events against a local cache keyed by UID + LAST-MODIFIED, then PUT new/changed events to a second calendar via calendar.save_event(). Guard against loops by tracking a X-SYNC-ORIGIN property.
• Structured backends. Replace write_output with a SQLite sink (sqlite3) or a Postgres insert to enable querying, full-text search on description, and incremental syncs based on last_modified.
• Prometheus metrics. Expose event counts, fetch duration, and error totals via prometheus_client in a long-running --watch mode that polls every N minutes — useful for alerting on stale calendars.
• OAuth2 / Bearer tokens. Swap DAVClient's Basic auth for a custom requests.Session with an Authorization: Bearer header, enabling Google Calendar and Microsoft Graph CalDAV bridges that reject app passwords.