📝 Add docstrings to codex/implement-circuit-breaker-and-hygraph-client-m5f9cc

Docstrings generation was requested by @shayancoin.

* #124 (comment)

The following files were modified:

* `backend/api/routes_sync.py`
* `backend/services/cache.py`
* `backend/services/circuit_breaker.py`
* `backend/services/hygraph_client.py`
* `backend/services/hygraph_service.py`

Author: coderabbitai[bot]

backend/api/routes_sync.py

@@ -47,12 +47,31 @@ def get_hygraph_service() -> HygraphService:
 
 
 def _error_envelope(code: str, message: str, details: Optional[dict] = None) -> Dict[str, Any]:
-    """Build a standardized error envelope for API responses."""
+    """
+    Create a standardized error envelope used for API error responses.
+    
+    Parameters:
+        code (str): Machine-readable error code.
+        message (str): Human-readable error message.
+        details (Optional[dict]): Additional structured error details; if omitted, an empty dict is used.
+    
+    Returns:
+        Dict[str, Any]: A dictionary of the form {"ok": False, "error": {"code": code, "message": message, "details": details}}.
+    """
 
     return {"ok": False, "error": {"code": code, "message": message, "details": details or {}}}
 
 
 def _processed_value(payload: Any) -> int:
+    """
+    Normalize a Hygraph payload into an integer count of processed items.
+    
+    Parameters:
+        payload (Any): Either a numeric-like value or a dict containing a "processed" key; other shapes are accepted but treated as zero when not convertible.
+    
+    Returns:
+        int: Integer count extracted from the payload; returns 0 if the value is missing, falsy, or cannot be converted to an integer.
+    """
     if isinstance(payload, dict):
         value = payload.get("processed", 0)
     else:
@@ -73,7 +92,14 @@ async def hygraph_webhook(
     background: BackgroundTasks,
     db: Session = Depends(get_db),
 ) -> Dict[str, Any]:
-    """Webhook receiver for Hygraph change notifications."""
+    """
+    Receive Hygraph webhook notifications, persist a deduplicating sync event, and schedule background processing to pull and reconcile Hygraph data.
+    
+    Persists a SyncEvent to detect duplicates; if the event is a duplicate returns a 200 response indicating deduplication. If the webhook payload is invalid JSON raises a 400 error. On a new event, schedules a background task that performs a full Hygraph pull and updates sync counters (or replays a circuit-breaker fallback payload if the pull is unavailable).
+    
+    Returns:
+    	JSONResponse: On duplicate events `{"ok": True, "dedup": True}` with status 200; on accepted new events `{"ok": True, "accepted": True}` with status 202.
+    """
 
     start = time.perf_counter()
     raw = getattr(request.state, "raw_body", b"")
@@ -109,6 +135,15 @@ async def hygraph_webhook(
 
 
     async def _process(event_id_local: Optional[str], body_sha_local: str) -> None:
+        """
+        Handle an asynchronous Hygraph sync triggered by a webhook: fetch counts, update Prometheus metrics, and log the outcome.
+        
+        If the pull succeeds, increments per-type upsert counters and a global success counter and emits a warning log with timing and counts. If a CircuitOpenError occurs, increments the circuit-breaker trip metric; if the exception provides a fallback payload, increments the fallback metric, replays its counts into the upsert counters and logs fallback details; if no fallback is available, increments the global failure metric and logs that the circuit is open. Any other exception increments the global failure metric and logs the error.
+        
+        Parameters:
+            event_id_local (Optional[str]): Identifier of the originating sync event (may be None).
+            body_sha_local (str): SHA256 of the webhook body used for traceability.
+        """
         t0 = time.perf_counter()
         try:
             counts = await HygraphService.pull_all(db)
@@ -172,7 +207,23 @@ async def hygraph_pull(
     body: Dict[str, Any] = Body(...),
     db: Session = Depends(get_db),
 ) -> Dict[str, Any]:
-    """Admin-triggered Hygraph syncs supporting manual pull types."""
+    """
+    Trigger an ad-hoc Hygraph sync for a specified resource type.
+    
+    Parameters:
+        body (dict): Request body that must include either `type` or `sync_type` (case-insensitive string) identifying the sync target:
+            - "materials", "modules", "systems", or "all".
+            Optionally accepts `page_size` (positive integer) to limit page size for the pull.
+        db (Session): Database session.
+    
+    Returns:
+        dict: {"ok": True, "data": counts} where `counts` is the result of the pull operation — either a mapping of resource types to payloads/counts (for "all") or a payload/count for the requested type.
+    
+    Raises:
+        HTTPException (400): when `page_size` is invalid or `type` is unsupported.
+        HTTPException (503): when the Hygraph circuit breaker is open and no fallback data is available.
+        HTTPException (500): on other internal sync failures.
+    """
 
     sync_type = str((body.get("type") or body.get("sync_type") or "")).lower().strip()
     page_size_raw = body.get("page_size")
@@ -236,4 +287,4 @@ async def hygraph_pull(
         logger.exception("hygraph_pull_failure", extra={"type": sync_type, "error": str(e)})
         raise HTTPException(status_code=500, detail=_error_envelope("INTERNAL", "sync failed", {"type": sync_type}))
 
-    return {"ok": True, "data": counts}
+    return {"ok": True, "data": counts}

backend/services/cache.py

@@ -26,6 +26,17 @@ class RedisCache:
     """Tiny wrapper that abstracts Redis availability."""
 
     def __init__(self, url: str | None = None, *, namespace: str = "hygraph") -> None:
+        """
+        Initialize the cache, preferring a Redis client when a URL is available and falling back to an in-memory store.
+        
+        Parameters:
+            url (str | None): Optional Redis connection URL; if not provided, the constructor will read HYGRAPH_CACHE_URL then REDIS_URL from the environment.
+            namespace (str): Key prefix used to namespace all cache entries (default "hygraph").
+        
+        Notes:
+            - Creates a thread-safe in-memory store and associated lock for the fallback path.
+            - When a Redis URL is available and the optional redis dependency is present, a Redis client is created and used for subsequent operations; otherwise the cache operates purely in-memory.
+        """
         self._url = url or os.getenv("HYGRAPH_CACHE_URL") or os.getenv("REDIS_URL")
         self._namespace = namespace
         self._lock = threading.Lock()
@@ -36,9 +47,28 @@ def __init__(self, url: str | None = None, *, namespace: str = "hygraph") -> Non
             self._client = None
 
     def _ns(self, key: str) -> str:
+        """
+        Compute the namespaced cache key.
+        
+        Prefixes the provided key with the instance namespace and a colon.
+        
+        Parameters:
+        	key (str): The cache key to namespace (without namespace prefix).
+        
+        Returns:
+        	namespaced_key (str): The key prefixed with "<namespace>:".
+        """
         return f"{self._namespace}:{key}"
 
     def get(self, key: str) -> Any | None:
+        """
+        Retrieve a cached value for the given key, preferring Redis and falling back to the in-memory store.
+        
+        If an in-memory entry has expired it is removed and treated as missing; Redis failures cause the client to be disabled and the in-memory store to be used thereafter.
+        
+        Returns:
+            The decoded Python object stored for the key, or `None` if not present or expired.
+        """
         namespaced = self._ns(key)
         if self._client is not None:
             try:
@@ -58,6 +88,16 @@ def get(self, key: str) -> Any | None:
             return json.loads(raw)
 
     def set(self, key: str, value: Any, *, ttl: int | None = None) -> None:
+        """
+        Store a value in the cache under the given key, preferring Redis and falling back to an in-memory store.
+        
+        The value is serialized to JSON before storage. If a Redis client is available, the entry is written to Redis (using expiry when `ttl` is provided). On Redis errors the instance disables the Redis client and stores the entry in the local in-memory store with an expiration timestamp computed from `ttl`. Keys are stored using the instance namespace prefix.
+        
+        Parameters:
+            key (str): Cache key (will be namespaced).
+            value (Any): Value to store; will be JSON-serialized.
+            ttl (int | None): Optional time-to-live in seconds; when provided, the entry expires after `ttl` seconds.
+        """
         payload = json.dumps(value)
         namespaced = self._ns(key)
         if self._client is not None:
@@ -75,6 +115,12 @@ def set(self, key: str, value: Any, *, ttl: int | None = None) -> None:
             self._memory[namespaced] = (payload, expires_at)
 
     def clear(self, key: str) -> None:
+        """
+        Remove the cached entry for the given key from Redis if available; otherwise remove it from the in-memory fallback.
+        
+        Parameters:
+            key (str): The cache key (un-prefixed) to remove.
+        """
         namespaced = self._ns(key)
         if self._client is not None:
             try:
@@ -84,4 +130,4 @@ def clear(self, key: str) -> None:
             else:
                 return
         with self._lock:
-            self._memory.pop(namespaced, None)
+            self._memory.pop(namespaced, None)

backend/services/circuit_breaker.py

@@ -14,6 +14,13 @@ class CircuitOpenError(RuntimeError):
     """Raised when the circuit breaker refuses to execute a call."""
 
     def __init__(self, message: str = "Circuit breaker is open", *, fallback: Any | None = None) -> None:
+        """
+        Create a CircuitOpenError carrying an optional fallback value.
+        
+        Parameters:
+            message (str): Error message describing the open-circuit condition.
+            fallback (Any | None): Optional value to be used as a fallback when the circuit is open; stored on the exception instance as `fallback`.
+        """
         super().__init__(message)
         self.fallback = fallback
 
@@ -35,7 +42,23 @@ class CircuitBreaker:
     HALF_OPEN: str = field(default="HALF_OPEN", init=False)
 
     def call(self, func: Callable[..., T], *args: Any, **kwargs: Any) -> T:
-        """Execute ``func`` respecting the breaker state machine."""
+        """
+        Execute a callable while enforcing the circuit breaker's state transitions.
+        
+        If the breaker is OPEN and the recovery timeout has not elapsed, a CircuitOpenError is raised; otherwise the breaker may transition to HALF_OPEN before executing the callable. Any exception raised by `func` is recorded as a failure and re-raised; a successful call is recorded as a success.
+        
+        Parameters:
+            func (Callable[..., T]): The callable to invoke.
+            *args: Positional arguments forwarded to `func`.
+            **kwargs: Keyword arguments forwarded to `func`.
+        
+        Returns:
+            T: The value returned by `func`.
+        
+        Raises:
+            CircuitOpenError: If the circuit is OPEN and not yet ready for recovery.
+            Exception: Re-raises any exception thrown by `func`.
+        """
 
         now = time.monotonic()
         if self.state == self.OPEN:
@@ -54,6 +77,11 @@ def call(self, func: Callable[..., T], *args: Any, **kwargs: Any) -> T:
         return result
 
     def _record_failure(self) -> None:
+        """
+        Record a failed call and update the circuit breaker's state.
+        
+        If the circuit is in HALF_OPEN, immediately move it to OPEN and record the failure time. Otherwise, increment the consecutive failure count and, when that count reaches or exceeds `failure_threshold`, move the circuit to OPEN and record the failure time.
+        """
         now = time.monotonic()
         if self.state == self.HALF_OPEN:
             self._trip(now)
@@ -64,6 +92,11 @@ def _record_failure(self) -> None:
             self._trip(now)
 
     def _record_success(self) -> None:
+        """
+        Record a successful call and update the circuit state accordingly.
+        
+        If the breaker is in HALF_OPEN, increment the consecutive success counter and close the circuit (clearing counters and last-failure timestamp) when the counter reaches or exceeds the configured success threshold. If the breaker is in any other state, close the circuit and clear counters and last-failure timestamp immediately.
+        """
         if self.state == self.HALF_OPEN:
             self.success_count += 1
             if self.success_count >= self.success_threshold:
@@ -72,13 +105,24 @@ def _record_success(self) -> None:
             self._reset()
 
     def _trip(self, when: float) -> None:
+        """
+        Transition the circuit to OPEN, record the trip timestamp, and reset failure/success counters.
+        
+        Parameters:
+            when (float): Monotonic timestamp indicating when the circuit was tripped.
+        """
         self.state = self.OPEN
         self.last_failure_time = when
         self.failure_count = 0
         self.success_count = 0
 
     def _reset(self) -> None:
+        """
+        Reset the circuit breaker to the CLOSED state and clear its counters and last failure timestamp.
+        
+        Sets `state` to `CLOSED`, sets `failure_count` and `success_count` to 0, and sets `last_failure_time` to `None`.
+        """
         self.state = self.CLOSED
         self.failure_count = 0
         self.success_count = 0
-        self.last_failure_time = None
+        self.last_failure_time = None

backend/services/hygraph_client.py

@@ -21,12 +21,27 @@ def __init__(
         breaker: CircuitBreaker | None = None,
         client: httpx.Client | None = None,
     ) -> None:
+        """
+        Initialize the HygraphClient with its GraphQL endpoint, authentication token, HTTP client, and circuit-breaker.
+        
+        Parameters:
+        	endpoint (str): GraphQL endpoint URL (must be provided as a keyword argument).
+        	token (str | None): Optional bearer token; `None` is treated as an empty string (no Authorization header).
+        	timeout (float): Default HTTP client timeout in seconds; used only when `client` is not provided.
+        	breaker (CircuitBreaker | None): Optional CircuitBreaker instance; a new CircuitBreaker is created if omitted.
+        	client (httpx.Client | None): Optional preconfigured httpx.Client; when provided, `timeout` is ignored.
+        """
         self._endpoint = endpoint
         self._token = token or ""
         self._breaker = breaker or CircuitBreaker()
         self._client = client or httpx.Client(timeout=timeout)
 
     def close(self) -> None:
+        """
+        Close the underlying HTTP client.
+        
+        Closes the internal httpx.Client and releases its associated network resources.
+        """
         self._client.close()
 
     def execute(
@@ -36,9 +51,33 @@ def execute(
         variables: Optional[Dict[str, Any]] = None,
         fallback: Callable[[], Any] | None = None,
     ) -> Dict[str, Any]:
-        """Execute a GraphQL query and return the ``data`` payload."""
+        """
+        Execute the GraphQL `query` and return its `data` payload.
+        
+        Parameters:
+            query (str): The GraphQL query string.
+            variables (Optional[Dict[str, Any]]): Variables for the GraphQL query; defaults to an empty dict when omitted.
+            fallback (Optional[Callable[[], Any]]): Callable invoked only when the circuit is open to produce a fallback payload; exceptions raised by the fallback are ignored and treated as no fallback.
+        
+        Returns:
+            Dict[str, Any]: The `data` field from the GraphQL response.
+        
+        Raises:
+            RuntimeError: If the GraphQL response contains `errors` or the `data` field is missing or not a dict.
+            CircuitOpenError: If the circuit breaker is open; the exception will include the fallback payload (if any).
+        """
 
         def _do_request() -> Dict[str, Any]:
+            """
+            Perform the GraphQL HTTP POST to the configured Hygraph endpoint and return the parsed `data` object.
+            
+            Returns:
+                dict: The GraphQL `data` payload.
+            
+            Raises:
+                httpx.HTTPStatusError: If the HTTP response status indicates failure.
+                RuntimeError: If the GraphQL response contains an `errors` entry, or if the `data` field is missing or not a dictionary.
+            """
             payload = {"query": query, "variables": variables or {}}
             headers = {"Content-Type": "application/json"}
             if self._token:
@@ -66,4 +105,10 @@ def _do_request() -> Dict[str, Any]:
 
     @property
     def breaker(self) -> CircuitBreaker:
-        return self._breaker
+        """
+        Expose the client's circuit breaker instance.
+        
+        Returns:
+            CircuitBreaker: The internal CircuitBreaker used to guard and monitor HTTP requests.
+        """
+        return self._breaker