redmine/redmine_crm/redmine_client.py

"""Redmine Contacts REST API client with retry, auth, and logging."""

from __future__ import annotations

import logging
import time
from dataclasses import dataclass
from http import HTTPStatus
from typing import Any, Dict, List, Optional

import requests

logger = logging.getLogger(__name__)


class RedmineError(Exception):
    """Base exception for Redmine client errors."""

    def __init__(self, message: str, *, redmine_id: Optional[int] = None) -> None:
        super().__init__(message)
        self.redmine_id = redmine_id


class RedmineAuthError(RedmineError):
    """Permanent auth failure (401, 403)."""


class RedmineTransientError(RedmineError):
    """Retryable error (network, 5xx, rate limit)."""


class RedmineNotFoundError(RedmineError):
    """Resource not found (404)."""


@dataclass(frozen=True, slots=True)
class RetryConfig:
    max_attempts: int = 5
    base_delay_seconds: float = 1.0
    max_delay_seconds: float = 30.0


class RedmineClient:
    """Thin, retry-aware wrapper around Redmine's REST API."""

    def __init__(
        self,
        base_url: str,
        api_key: str,
        *,
        project_id: Optional[int] = None,
        timeout: float = 30.0,
        retry: Optional[RetryConfig] = None,
    ) -> None:
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key
        self.project_id = project_id
        self.timeout = timeout
        self.retry = retry or RetryConfig()
        self.session = requests.Session()
        self.session.headers.update({
            "X-Redmine-API-Key": self.api_key,
            "Content-Type": "application/json",
            "Accept": "application/json",
        })

    def _request(
        self,
        method: str,
        path: str,
        *,
        params: Optional[Dict[str, Any]] = None,
        json_body: Optional[Dict[str, Any]] = None,
        redmine_id: Optional[int] = None,
    ) -> Dict[str, Any]:
        url = f"{self.base_url}{path}"
        last_error: Optional[Exception] = None

        for attempt in range(1, self.retry.max_attempts + 1):
            transient_error: Optional[RedmineTransientError] = None

            try:
                response = self.session.request(
                    method,
                    url,
                    params=params,
                    json=json_body,
                    timeout=self.timeout,
                )
            except requests.RequestException as exc:
                last_error = exc
                logger.warning(
                    "Redmine network error on attempt %d/%d: %s",
                    attempt,
                    self.retry.max_attempts,
                    exc,
                )
                if attempt == self.retry.max_attempts:
                    raise RedmineTransientError(
                        f"Network error after {self.retry.max_attempts} attempts: {exc}",
                        redmine_id=redmine_id,
                    ) from exc
            else:
                logger.debug(
                    "Redmine request %s %s attempt %d/%d status=%d",
                    method,
                    url,
                    attempt,
                    self.retry.max_attempts,
                    response.status_code,
                )

                if response.status_code == HTTPStatus.OK.value:
                    if not response.content:
                        return {}
                    return response.json()

                if response.status_code in (
                    HTTPStatus.UNAUTHORIZED.value,
                    HTTPStatus.FORBIDDEN.value,
                ):
                    raise RedmineAuthError(
                        f"Authentication failed ({response.status_code})",
                        redmine_id=redmine_id,
                    )

                if response.status_code == HTTPStatus.NOT_FOUND.value:
                    raise RedmineNotFoundError(
                        f"Not found ({response.status_code})",
                        redmine_id=redmine_id,
                    )

                if response.status_code == HTTPStatus.TOO_MANY_REQUESTS.value:
                    transient_error = RedmineTransientError(
                        "Rate limited (429)", redmine_id=redmine_id
                    )
                elif response.status_code >= HTTPStatus.INTERNAL_SERVER_ERROR.value:
                    transient_error = RedmineTransientError(
                        f"Server error ({response.status_code})",
                        redmine_id=redmine_id,
                    )
                else:
                    # 4xx other than auth/404 are permanent.
                    raise RedmineError(
                        f"Client error ({response.status_code}): {response.text}",
                        redmine_id=redmine_id,
                    )

            if transient_error is not None:
                if attempt == self.retry.max_attempts:
                    raise transient_error
                logger.warning(
                    "Redmine transient error on attempt %d/%d: %s",
                    attempt,
                    self.retry.max_attempts,
                    transient_error,
                )

            if attempt < self.retry.max_attempts:
                delay = min(
                    self.retry.base_delay_seconds * (2 ** (attempt - 1)),
                    self.retry.max_delay_seconds,
                )
                logger.info("Retrying Redmine request in %.1f seconds", delay)
                time.sleep(delay)

        # Should never be reached because the loop raises.
        raise RedmineTransientError("Unexpected loop exit", redmine_id=redmine_id) from last_error

    def fetch_contacts(
        self,
        *,
        updated_since: Optional[str] = None,
        limit: int = 100,
        offset: int = 0,
        status_filter: Optional[str] = "active",
    ) -> List[Dict[str, Any]]:
        params: Dict[str, Any] = {"limit": limit, "offset": offset}
        if self.project_id is not None:
            params["project_id"] = self.project_id
        if updated_since:
            params["updated_on"] = f">={updated_since}"
        if status_filter:
            params["status"] = status_filter

        data = self._request("GET", "/contacts.json", params=params)
        contacts = data.get("contacts", [])
        logger.info(
            "Fetched %d contacts from Redmine (limit=%d, offset=%d)",
            len(contacts),
            limit,
            offset,
        )
        return contacts

    def fetch_contact(self, contact_id: int) -> Dict[str, Any]:
        data = self._request(
            "GET",
            f"/contacts/{contact_id}.json",
            redmine_id=contact_id,
        )
        return data.get("contact", data)

    def iter_all_contacts(
        self,
        *,
        updated_since: Optional[str] = None,
        batch_size: int = 100,
    ) -> List[Dict[str, Any]]:
        """Iterate through paginated contacts and collect all results."""
        all_contacts: List[Dict[str, Any]] = []
        offset = 0
        while True:
            batch = self.fetch_contacts(
                updated_since=updated_since,
                limit=batch_size,
                offset=offset,
            )
            if not batch:
                break
            all_contacts.extend(batch)
            if len(batch) < batch_size:
                break
            offset += batch_size
        return all_contacts