refactor: extract rate-limit helpers; add async rate-limit support

Move Retry-After parsing and sleep-decision logic into three shared
module-level helpers in caldav/lib/error.py, adjacent to RateLimitError:

  parse_retry_after(header)             -> float | None
  compute_sleep_seconds(s, default, max) -> float | None
  raise_if_rate_limited(status, url, header) -> None (raises on 429/503)

Both sync (DAVClient) and async (AsyncDAVClient) clients now call these
helpers instead of duplicating the logic inline.

Changes per file:
- caldav/lib/error.py: add helpers + necessary imports (datetime, timezone,
  parsedate_to_datetime)
- caldav/davclient.py: replace 15-line inline parse block with single
  raise_if_rate_limited() call; simplify request() except handler via
  compute_sleep_seconds(); remove now-unused datetime/timezone/
  parsedate_to_datetime imports
- caldav/async_davclient.py: add rate_limit_handle/default_sleep/max_sleep
  params to __init__; split monolithic request() into request() (thin
  rate-limit wrapper using asyncio.sleep) + _async_request() (HTTP call
  + auth negotiation); insert raise_if_rate_limited() call in _async_request
- tests/test_caldav_unit.py: add TestRateLimitHelpers (17 tests covering
  all three helpers directly)
- tests/test_async_davclient.py: add TestAsyncRateLimiting (9 tests
  mirroring the sync TestRateLimiting class, using AsyncMock + patched
  asyncio.sleep)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: tobixen

caldav/async_davclient.py

@@ -6,6 +6,7 @@
 For sync usage, see the davclient.py wrapper.
 """
 
+import asyncio
 import logging
 import sys
 from collections.abc import Mapping
@@ -143,6 +144,9 @@ def __init__(
         features: FeatureSet | dict | str | None = None,
         enable_rfc6764: bool = True,
         require_tls: bool = True,
+        rate_limit_handle: bool = False,
+        rate_limit_default_sleep: Optional[int] = None,
+        rate_limit_max_sleep: Optional[int] = None,
     ) -> None:
         """
         Initialize an async DAV client.
@@ -162,6 +166,13 @@ def __init__(
             features: FeatureSet for server compatibility workarounds.
             enable_rfc6764: Enable RFC6764 DNS-based service discovery.
             require_tls: Require TLS for discovered services (security consideration).
+            rate_limit_handle: When True, automatically sleep and retry on 429/503
+                responses. When False (default), raise RateLimitError immediately.
+            rate_limit_default_sleep: Fallback sleep seconds when the server's 429
+                response omits a Retry-After header. None (default) means raise
+                rather than sleeping when no Retry-After is provided.
+            rate_limit_max_sleep: Cap on sleep duration in seconds regardless of
+                server's Retry-After value. None (default) means no cap.
         """
         headers = headers or {}
 
@@ -247,6 +258,10 @@ def __init__(
         }
         self.headers.update(headers)
 
+        self.rate_limit_handle = rate_limit_handle
+        self.rate_limit_default_sleep = rate_limit_default_sleep
+        self.rate_limit_max_sleep = rate_limit_max_sleep
+
     def _create_session(self) -> None:
         """Create or recreate the async HTTP client with current settings."""
         if _USE_HTTPX:
@@ -326,16 +341,39 @@ async def request(
         headers: Mapping[str, str] | None = None,
     ) -> AsyncDAVResponse:
         """
-        Send an async HTTP request.
+        Send an async HTTP request, with optional rate-limit sleep-and-retry.
 
-        Args:
-            url: Request URL.
-            method: HTTP method.
-            body: Request body.
-            headers: Additional headers.
+        Catches RateLimitError from _async_request. When rate_limit_handle is
+        True and a usable sleep duration is available, sleeps then retries once.
+        Otherwise re-raises immediately.
+        """
+        try:
+            return await self._async_request(url, method, body, headers)
+        except error.RateLimitError as e:
+            if not self.rate_limit_handle:
+                raise
+            sleep_seconds = error.compute_sleep_seconds(
+                e.retry_after_seconds,
+                self.rate_limit_default_sleep,
+                self.rate_limit_max_sleep,
+            )
+            if sleep_seconds is None:
+                raise
+            await asyncio.sleep(sleep_seconds)
+            return await self._async_request(url, method, body, headers)
 
-        Returns:
-            AsyncDAVResponse object.
+    async def _async_request(
+        self,
+        url: str,
+        method: str = "GET",
+        body: str = "",
+        headers: Mapping[str, str] | None = None,
+    ) -> AsyncDAVResponse:
+        """
+        Async HTTP request implementation with auth negotiation.
+
+        Handles connection-abort workaround, 429/503 rate-limit detection,
+        and 401 auth negotiation (including HTTP/2 fallback).
         """
         headers = headers or {}
 
@@ -438,8 +476,10 @@ async def request(
                 r = await self.session.request(**request_kwargs)
             response = AsyncDAVResponse(r, self)
 
-        # Handle 401 responses for auth negotiation (after try/except)
-        # This matches the original sync client's auth negotiation logic
+        # Handle 429/503 rate-limit responses
+        error.raise_if_rate_limited(r.status_code, str(url_obj), r.headers.get("Retry-After"))
+
+        # Handle 401 responses for auth negotiation
         # httpx headers are already case-insensitive
         if (
             r.status_code == 401
@@ -458,7 +498,7 @@ async def request(
                 )
 
             # Retry request with authentication
-            return await self.request(url, method, body, headers)
+            return await self._async_request(url, method, body, headers)
 
         elif (
             r.status_code == 401
@@ -467,36 +507,28 @@ async def request(
             and self.password
             and isinstance(self.password, bytes)
         ):
-            # Handle HTTP/2 issue (matches original sync client)
-            # Most likely wrong username/password combo, but could be an HTTP/2 problem
+            # Handle HTTP/2 issue: most likely wrong username/password, but could be an
+            # HTTP/2 problem.  Retry with HTTP/2 disabled if multiplexing was auto-detected.
             if self.features.is_supported("http.multiplexing", return_defaults=False) is None:
-                await self.close()  # Uses correct close method for httpx/niquests
+                await self.close()
                 self._http2 = False
                 self._create_session()
                 # Set multiplexing to False BEFORE retry to prevent infinite loop
-                # If the retry succeeds, this was the right choice
-                # If it also fails with 401, it's not a multiplexing issue but an auth issue
                 self.features.set_feature("http.multiplexing", False)
-                # If this one also fails, we give up
-                ret = await self.request(str(url_obj), method, body, headers)
-                return ret
-
-            # Most likely we're here due to wrong username/password combo,
-            # but it could also be charset problems. Some (ancient) servers
-            # don't like UTF-8 binary auth with Digest authentication.
-            # An example are old SabreDAV based servers. Not sure about UTF-8
-            # and Basic Auth, but likely the same. So retry if password is
-            # a bytes sequence and not a string.
+                return await self._async_request(str(url_obj), method, body, headers)
+
+            # Could also be charset problems with old SabreDAV-based servers that don't like
+            # UTF-8 binary auth with Digest/Basic authentication.
             auth_types = self.extract_auth_types(r.headers["WWW-Authenticate"])
             self.password = self.password.decode()
             self.build_auth_object(auth_types)
 
             self.username = None
             self.password = None
 
-            return await self.request(str(url_obj), method, body, headers)
+            return await self._async_request(str(url_obj), method, body, headers)
 
-        # Raise AuthorizationError for 401/403 responses (matches original sync client)
+        # Raise AuthorizationError for 401/403 responses
         if response.status in (401, 403):
             try:
                 reason = response.reason

caldav/davclient.py

@@ -13,8 +13,6 @@
 import sys
 import time
 import warnings
-from datetime import datetime, timezone
-from email.utils import parsedate_to_datetime
 from types import TracebackType
 from typing import TYPE_CHECKING, Any, Optional
 from urllib.parse import unquote
@@ -961,18 +959,14 @@ def request(
         except error.RateLimitError as e:
             if not self.rate_limit_handle:
                 raise
-            retry_after_seconds = (
-                e.retry_after_seconds
-                if e.retry_after_seconds is not None
-                else self.rate_limit_default_sleep
+            sleep_seconds = error.compute_sleep_seconds(
+                e.retry_after_seconds,
+                self.rate_limit_default_sleep,
+                self.rate_limit_max_sleep,
             )
-            if retry_after_seconds is None or retry_after_seconds <= 0:
+            if sleep_seconds is None:
                 raise
-            if self.rate_limit_max_sleep is not None:
-                retry_after_seconds = min(retry_after_seconds, self.rate_limit_max_sleep)
-            if retry_after_seconds <= 0:
-                raise
-            time.sleep(retry_after_seconds)
+            time.sleep(sleep_seconds)
             return self._sync_request(url, method, body, headers)
 
     def _sync_request(
@@ -1019,26 +1013,7 @@ def _sync_request(
         r_headers = CaseInsensitiveDict(r.headers)
 
         # Handle 429/503 responses: raise RateLimitError so the caller can decide whether to retry
-        if r.status_code in (429, 503):
-            retry_after_header: Optional[str] = r_headers.get("Retry-After")
-            retry_seconds: Optional[float] = None
-            if retry_after_header:
-                try:
-                    retry_seconds = int(retry_after_header)
-                except ValueError:
-                    try:
-                        retry_date = parsedate_to_datetime(retry_after_header)
-                        now = datetime.now(timezone.utc)
-                        retry_seconds = max(0.0, (retry_date - now).total_seconds())
-                    except (ValueError, TypeError):
-                        pass
-            if r.status_code == 429 or retry_after_header is not None:
-                raise error.RateLimitError(
-                    url=str(url_obj),
-                    reason=f"Rate limited or service unavailable. Retry after: {retry_after_header}",
-                    retry_after=retry_after_header,
-                    retry_after_seconds=retry_seconds,
-                )
+        error.raise_if_rate_limited(r.status_code, str(url_obj), r_headers.get("Retry-After"))
 
         # Handle 401 responses for auth negotiation
         if (

caldav/lib/error.py

@@ -1,6 +1,9 @@
 #!/usr/bin/env python
 import logging
 from collections import defaultdict
+from datetime import datetime, timezone
+from email.utils import parsedate_to_datetime
+from typing import Optional
 
 from caldav import __version__
 
@@ -149,6 +152,85 @@ def __init__(
         self.retry_after_seconds = retry_after_seconds
 
 
+def parse_retry_after(retry_after_header: str) -> Optional[float]:
+    """Parse a Retry-After header value into seconds from now.
+
+    Handles both the integer-seconds form (RFC 7231 §7.1.3) and the HTTP-date
+    form.  Returns None if the value cannot be parsed.
+    """
+    try:
+        return float(int(retry_after_header))
+    except ValueError:
+        pass
+    try:
+        retry_date = parsedate_to_datetime(retry_after_header)
+        now = datetime.now(timezone.utc)
+        return max(0.0, (retry_date - now).total_seconds())
+    except (ValueError, TypeError):
+        return None
+
+
+def compute_sleep_seconds(
+    retry_after_seconds: Optional[float],
+    default_sleep: Optional[int],
+    max_sleep: Optional[int],
+) -> Optional[float]:
+    """Compute the effective sleep duration for rate-limit handling.
+
+    Returns None when there is no usable duration (meaning the caller should
+    re-raise the RateLimitError rather than sleeping).
+
+    Args:
+        retry_after_seconds: Parsed seconds from server Retry-After header,
+            or None if the server did not provide one.
+        default_sleep: Fallback duration when retry_after_seconds is None.
+            None means no fallback; re-raise.
+        max_sleep: Hard cap on sleep duration.  None means no cap; 0 means
+            never sleep.
+    """
+    effective: Optional[float] = (
+        retry_after_seconds
+        if retry_after_seconds is not None
+        else (float(default_sleep) if default_sleep is not None else None)
+    )
+    if effective is None or effective <= 0:
+        return None
+    if max_sleep is not None:
+        effective = min(effective, float(max_sleep))
+    if effective <= 0:
+        return None
+    return effective
+
+
+def raise_if_rate_limited(
+    status_code: int,
+    url: str,
+    retry_after_header: Optional[str],
+) -> None:
+    """Raise RateLimitError when the response indicates rate limiting.
+
+    Raises for:
+    - Any 429 response (regardless of Retry-After presence)
+    - 503 responses that include a Retry-After header
+
+    Args:
+        status_code: HTTP status code from the response.
+        url: Request URL (included in exception for context).
+        retry_after_header: Raw Retry-After header value, or None.
+    """
+    if status_code not in (429, 503):
+        return
+    if status_code != 429 and retry_after_header is None:
+        return
+    retry_seconds = parse_retry_after(retry_after_header) if retry_after_header else None
+    raise RateLimitError(
+        url=url,
+        reason=f"Rate limited or service unavailable. Retry after: {retry_after_header}",
+        retry_after=retry_after_header,
+        retry_after_seconds=retry_seconds,
+    )
+
+
 exception_by_method: dict[str, DAVError] = defaultdict(lambda: DAVError)
 for method in (
     "delete",