"""Dependency-free RideWrench fleet and provider API client.""" from dataclasses import dataclass import hashlib import hmac import json import secrets import time from typing import Any, Dict, Generator, Optional from urllib.error import HTTPError from urllib.parse import urlencode from urllib.request import Request, urlopen @dataclass class RideWrenchResponse: data: Any next_cursor: Optional[str] request_id: Optional[str] rate_limit: Optional[int] rate_limit_remaining: Optional[int] idempotent_replay: bool class RideWrenchError(Exception): def __init__(self, message: str, *, status: int, code: Optional[str] = None, details: Any = None, request_id: Optional[str] = None, retry_after: Optional[str] = None, operation_unconfirmed: bool = False): super().__init__(message) self.status = status self.code = code self.details = details self.request_id = request_id self.retry_after = retry_after self.operation_unconfirmed = operation_unconfirmed class RideWrench: def __init__(self, *, api_key: str, audience: str, base_url: str = "https://www.ridewrench.com", timeout: int = 30): if not api_key: raise ValueError("api_key is required") if audience not in ("fleet", "provider"): raise ValueError("audience must be 'fleet' or 'provider'") self.api_key = api_key self.audience = audience self.base_url = base_url.rstrip("/") self.timeout = timeout self.prefix = "/api/v1/provider" if audience == "provider" else "/api/v1" self.fleet = FleetAPI(self) self.provider = ProviderAPI(self) def request(self, method: str, path: str, *, body: Optional[Dict[str, Any]] = None, query: Optional[Dict[str, Any]] = None, idempotency_key: Optional[str] = None) -> RideWrenchResponse: method = method.upper() write = method not in ("GET", "HEAD", "OPTIONS") values = {key: value for key, value in (query or {}).items() if value is not None and value != ""} url = f"{self.base_url}{path}" + (f"?{urlencode(values)}" if values else "") encoded = json.dumps(body).encode() if body is not None else None headers = {"Authorization": f"Bearer {self.api_key}", "Accept": "application/json"} if encoded is not None: headers["Content-Type"] = "application/json" if write: headers["Idempotency-Key"] = idempotency_key or secrets.token_urlsafe(24) request = Request(url, data=encoded, headers=headers, method=method) try: response = urlopen(request, timeout=self.timeout) status, response_headers, raw = response.status, response.headers, response.read() except HTTPError as error: status, response_headers, raw = error.code, error.headers, error.read() try: payload = json.loads(raw or b"{}") except json.JSONDecodeError: payload = {} request_id = response_headers.get("X-Request-Id") or payload.get("requestId") if status < 200 or status >= 300: error = payload.get("error") or {} raise RideWrenchError( error.get("message") or f"RideWrench request failed with status {status}.", status=status, code=error.get("code"), details=error.get("details"), request_id=request_id, retry_after=response_headers.get("Retry-After"), operation_unconfirmed=write and status >= 500, ) return RideWrenchResponse( data=payload.get("data", payload), next_cursor=payload.get("nextCursor"), request_id=request_id, rate_limit=_integer(response_headers.get("X-RateLimit-Limit")), rate_limit_remaining=_integer(response_headers.get("X-RateLimit-Remaining")), idempotent_replay=response_headers.get("Idempotent-Replayed") == "true", ) def pages(self, path: str, *, limit: int = 50) -> Generator[RideWrenchResponse, None, None]: cursor = None while True: page = self.request("GET", path, query={"limit": limit, "cursor": cursor}) yield page cursor = page.next_cursor if not cursor: break def _integer(value: Optional[str]) -> Optional[int]: try: return int(value) if value is not None else None except ValueError: return None class Collection: def __init__(self, client: RideWrench, resource: str, creatable: bool = False, updateable: bool = False): self.client, self.resource, self.creatable, self.updateable = client, resource, creatable, updateable self.path = f"{client.prefix}/{resource}" def list(self, *, limit: int = 50, cursor: Optional[str] = None) -> RideWrenchResponse: return self.client.request("GET", self.path, query={"limit": limit, "cursor": cursor}) def get(self, resource_id: str) -> RideWrenchResponse: return self.client.request("GET", f"{self.path}/{resource_id}") def pages(self, *, limit: int = 50) -> Generator[RideWrenchResponse, None, None]: return self.client.pages(self.path, limit=limit) def create(self, body: Dict[str, Any], *, idempotency_key: Optional[str] = None) -> RideWrenchResponse: if not self.creatable: raise AttributeError("This collection does not support create") return self.client.request("POST", self.path, body=body, idempotency_key=idempotency_key) def update(self, resource_id: str, body: Dict[str, Any], *, idempotency_key: Optional[str] = None) -> RideWrenchResponse: if not self.updateable: raise AttributeError("This collection does not support update") return self.client.request("PATCH", f"{self.path}/{resource_id}", body=body, idempotency_key=idempotency_key) class FleetAPI: def __init__(self, client: RideWrench): self.client = client for resource in ("equipment", "breakdowns", "maintenance", "webhooks"): setattr(self, resource.replace("-", "_"), Collection(client, resource, True, True)) self.work_orders = Collection(client, "work-orders", False, True) self.dvir = Collection(client, "dvir") self.events = Collection(client, "events", True, False) def profile(self) -> RideWrenchResponse: return self.client.request("GET", f"{self.client.prefix}/fleet") def ingest_event(self, event: Dict[str, Any], *, idempotency_key: Optional[str] = None) -> RideWrenchResponse: return self.events.create(event, idempotency_key=idempotency_key) class ProviderAPI: def __init__(self, client: RideWrench): self.client = client for resource in ("locations", "technicians", "work-orders", "invoices", "payments", "webhooks"): setattr(self, resource.replace("-", "_"), Collection(client, resource)) for resource in ("customers", "assets", "appointments", "direct-work-orders", "catalog-items", "inventory-items", "reconciliation-entries"): setattr(self, resource.replace("-", "_"), Collection(client, resource, True, True)) def profile(self) -> RideWrenchResponse: return self.client.request("GET", self.client.prefix) def work_order_action(self, work_order_id: str, action: str, *, idempotency_key: Optional[str] = None, **values: Any) -> RideWrenchResponse: return self.client.request("POST", f"{self.client.prefix}/work-orders/{work_order_id}/actions", body={"action": action, **values}, idempotency_key=idempotency_key) def accept(self, work_order_id: str, **options: Any) -> RideWrenchResponse: return self.work_order_action(work_order_id, "accept", **options) def decline(self, work_order_id: str, reason: str, **options: Any) -> RideWrenchResponse: return self.work_order_action(work_order_id, "decline", reason=reason, **options) def assign_technician(self, work_order_id: str, technician_email: str, **options: Any) -> RideWrenchResponse: return self.work_order_action(work_order_id, "assign_technician", technicianEmail=technician_email, **options) def update_status(self, work_order_id: str, status: str, **options: Any) -> RideWrenchResponse: return self.work_order_action(work_order_id, "update_status", status=status, **options) def update_eta(self, work_order_id: str, eta_minutes: int, **options: Any) -> RideWrenchResponse: return self.work_order_action(work_order_id, "update_eta", etaMinutes=eta_minutes, **options) def submit_estimate(self, work_order_id: str, estimate: Dict[str, Any], **options: Any) -> RideWrenchResponse: return self.client.request("POST", f"{self.client.prefix}/work-orders/{work_order_id}/estimates", body=estimate, **options) def send_message(self, work_order_id: str, message: str, **options: Any) -> RideWrenchResponse: return self.client.request("POST", f"{self.client.prefix}/work-orders/{work_order_id}/messages", body={"message": message}, **options) def invoice(self, payment_case_id: str, action: str, invoice: Optional[Dict[str, Any]] = None, **options: Any) -> RideWrenchResponse: return self.client.request("POST", f"{self.client.prefix}/invoices/{payment_case_id}", body={**(invoice or {}), "action": action}, **options) def create_webhook(self, subscription: Dict[str, Any], **options: Any) -> RideWrenchResponse: return self.client.request("POST", f"{self.client.prefix}/webhooks", body=subscription, **options) def webhook_deliveries(self, webhook_id: str, *, limit: int = 50, cursor: Optional[str] = None) -> RideWrenchResponse: return self.client.request("GET", f"{self.client.prefix}/webhooks/{webhook_id}/deliveries", query={"limit": limit, "cursor": cursor}) def replay_webhook(self, webhook_id: str, delivery_id: str, **options: Any) -> RideWrenchResponse: return self.client.request("POST", f"{self.client.prefix}/webhooks/{webhook_id}/replay", body={"deliveryId": delivery_id}, **options) def verify_webhook_signature(*, secret: str, payload: bytes, signature: str, timestamp: str, tolerance_seconds: int = 300, now: Optional[float] = None) -> bool: try: if abs(int(now or time.time()) - int(timestamp)) > tolerance_seconds: return False except (TypeError, ValueError): return False supplied = signature[7:] if signature.startswith("sha256=") else signature expected = hmac.new(secret.encode(), timestamp.encode() + b"." + payload, hashlib.sha256).hexdigest() return len(supplied) == 64 and hmac.compare_digest(expected, supplied) def construct_webhook_event(**options: Any) -> Dict[str, Any]: if not verify_webhook_signature(**options): raise ValueError("Webhook signature is invalid or stale") return json.loads(options["payload"])