Fix SSE field casing and add event handler support#46
Merged
Conversation
The Risco cloud SSE API sends IsOffline and LastStatusUpdate in PascalCase, but the code was reading them as isOffline/lastStatusUpdate, causing the offline check and deduplication logic to silently do nothing. Also adds push-based event handling: when the SSE runtimeUpdate message includes LastEventUpdated, get_events is called and registered event handlers are invoked, so callers no longer need to poll for new events. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
There was a problem hiding this comment.
Pull request overview
This PR updates the Risco Cloud SSE integration to correctly parse the API’s PascalCase runtimeUpdate payload fields and adds push-based event-log notifications when LastEventUpdated is present.
Changes:
- Fix SSE payload field casing (
IsOffline,LastStatusUpdate) so offline guarding and status deduplication work correctly. - Add event-log handler support via
add_event_handler, tracking_last_event_updateand fetching events whenLastEventUpdatedadvances. - Update tests and the SSE example to reflect the real API casing and new event handler flow.
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
pyrisco/cloud/risco_cloud.py |
Fixes SSE field casing; introduces event handler registration and LastEventUpdated-driven event fetching. |
tests/test_risco_cloud.py |
Updates SSE test payload casing and adds tests covering the event-handler path. |
examples/cloud/cloud_sse_example.py |
Demonstrates add_event_handler usage in the SSE example. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
The API can return a null response body when the event log query returns nothing. Guard against this by returning an empty list instead of crashing. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Two issues from code review: - _last_event_update is a datetime; convert to isoformat() before passing as newerThan to avoid JSON serialization error on the second fetch - Skip get_events entirely when no event handlers are registered Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 3 out of 3 changed files in this pull request and generated 1 comment.
Comments suppressed due to low confidence (1)
pyrisco/cloud/risco_cloud.py:179
- Event updates are currently gated behind the
LastStatusUpdatededuplication return. If aruntimeUpdatearrives with a non-newerLastStatusUpdate(or one that’s missing) but a newerLastEventUpdated, this function will return early and never fetch/notify event handlers. Consider handlingLastEventUpdatedindependently of the state-update dedupe (e.g., evaluate event timestamp before theupdate_time <= _last_status_updateearly-return, and skip only the state fetch when status hasn’t changed).
ts_str = data.get("LastStatusUpdate")
if not ts_str:
return
update_time = datetime.fromisoformat(ts_str)
if self._last_status_update is not None and update_time <= self._last_status_update:
return
- Restructure _handle_runtime_update so LastEventUpdated is processed independently of LastStatusUpdate: a message with only LastEventUpdated (or a non-newer LastStatusUpdate) now correctly triggers event handlers - Replace `if not response` with `if response is None` in get_events so an unexpected empty-dict response surfaces as a KeyError rather than being silently swallowed Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Before consuming any SSE messages, do one state fetch and notify handlers. Since the SSE connection is already open at that point, the server is buffering events — so no state changes in the window between the fetch and the first message can be missed. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
… mismatch Storing the datetime and calling .isoformat() produces +00:00, inconsistent with the Z-suffixed strings the API uses. Store the original SSE string instead and pass it directly to get_events; parse to datetime only locally for comparison. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Network drops (e.g. ClientPayloadError mid-stream) are now handled internally: the loop catches the exception, calls error handlers for observability, sleeps RECONNECT_DELAY (30 s), then reconnects without requiring any action from the caller. The SSE example is updated to reflect that error handlers are now informational-only and no longer need to drive reconnection. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Replace the flat 30s delay with exponential backoff: 1s, 2s, 4s, 8s… capped at 60s. After RECONNECT_MAX_ATTEMPTS (5) consecutive failures the loop gives up and calls error handlers one final time with MaxRetriesError wrapping the last exception, then exits. A successful connection resets the attempt counter so transient drops don't accumulate toward the limit. MaxRetriesError is exported from the top-level pyrisco package. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
- Show add_event_handler usage - Remove manual reconnect from on_error (now handled automatically) - Show MaxRetriesError handling pattern for when retries are exhausted - Document exponential backoff and initial-state-on-subscribe behavior Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
close() now checks asyncio.current_task() before cancelling/awaiting _subscription_task, so calling close() from within the subscription task (e.g. via UnauthorizedError in _site_post) no longer deadlocks. Add a RECONNECT_INITIAL_DELAY sleep after clean stream EOF before the next connect attempt, preventing a tight reconnect loop when the server cycles connections normally. Mock asyncio.sleep at the TestRiscoCloud class level so reconnect delays never slow the test suite, regardless of which code path is exercised. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
- Add _parse_timestamp() helper that normalises trailing 'Z' to '+00:00' before calling fromisoformat(), making the code work on Python < 3.11 and removing the need for callers to know about the format difference - In README and example, schedule re-login via asyncio.create_task() so close() is called from a new task rather than from within the running SSE task — calling close() from within the task schedules CancelledError on itself, which would abort login() before it could complete Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
- Call resp.raise_for_status() right after opening the SSE connection so 4xx/5xx responses are treated as errors and go through the backoff/retry path rather than silently appearing as a clean EOF that resets attempt=0 - Revert close() to skip cancel() (not just await) when called from within the subscription task; calling cancel() on the current task schedules CancelledError at the next await, which can abort session.close() and other cleanup steps mid-flight. When close() is called from inside the task (e.g. UnauthorizedError recovery), the loop stops naturally because the session is closed beneath it. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
The real root cause of the close()-from-within-task problem was that _site_post's UnauthorizedError recovery called close() + login() from inside the SSE loop. close() cancels the subscription task, which means any subsequent await (including session.close()) could receive CancelledError and abort cleanup. Fix: extract _relogin() which refreshes credentials in-place (clears session_id, re-runs the three login steps) without touching the session or the SSE task. _site_post now calls _relogin() instead of close()+login() on UnauthorizedError. close() is now simple and correct: always cancel+await, no current_task check needed. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
The SSE stream drops for network reasons — the session token stays valid, so UnauthorizedError never fires from within _sse_loop in practice. All the _relogin() / close()-self-cancel complexity was solving a problem that doesn't occur. - Remove _relogin(); restore _site_post to close()+login() for the external arm/disarm case where 401 can legitimately happen - Fix attempt reset: move it back to after clean EOF (not after raise_for_status). With the early reset, post-connect failures (e.g. initial state fetch) always saw attempt=0 and MaxRetriesError could never be reached - Fix README and example comment: the real risk of calling close() from within the SSE error handler is a deadlock (close() awaits the running task itself), not that cancel() fires before login() completes Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
- Move attempt=0 to after initial state fetch succeeds (right before the message loop). Previously reset at clean EOF meant that if TCP connected but the state fetch always failed, attempt never accumulated past 1 and MaxRetriesError could never fire. Now reset requires the connection to reach a usable state (SSE up + initial state fetched). - Add current_task() guard in close() so it always calls cancel() (to stop the loop) but skips await when called from within the subscription task itself, preventing a self-await deadlock. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Add _call_handlers() static method that swallows per-handler exceptions so a buggy or raising callback doesn't propagate into the SSE loop and trigger a spurious reconnect/backoff. Pattern mirrors pyrisco/local's _call_handlers. All state/event handler invocations now go through it. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Switch _call_handlers to the same non-blocking dispatch as pyrisco/local/risco_local.py: a synchronous method that schedules asyncio.create_task(_gather()) so handler calls don't block the SSE reader. return_exceptions=True on gather means handler exceptions are captured, not propagated. Apply to error handlers too for consistency — a raising error handler no longer risks aborting the reconnect sleep or MaxRetriesError return. Tests drain pending handler tasks after the subscription task ends via asyncio.gather(*all_tasks) so assertions remain deterministic. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
isOffline→IsOffline,lastStatusUpdate→LastStatusUpdateso offline guard and deduplication actually workLastEventUpdated; when present and newer,get_eventsis called and registeredadd_event_handlercallbacks are notified (no more polling required)LastEventUpdatedindependent ofLastStatusUpdate— the two update paths are decoupled so a message carrying onlyLastEventUpdatedstill fires event handlersRECONNECT_MAX_ATTEMPTS(5) consecutive connection failures the error handler is called withMaxRetriesErrorand the loop exitsMaxRetriesError— new exception exported frompyrisco; wraps the last underlying error so callers can distinguish "still retrying" from "gave up". Recommended response is to schedule a re-login viaasyncio.create_task(not directly from the error handler, sinceclose()awaits the subscription task and would deadlock if called from within it)resp.raise_for_status()on connect so 4xx/5xx responses go through the backoff/retry path instead of appearing as a clean EOFMaxRetriesErrorhandling pattern,add_event_handler, and remove the manual reconnect boilerplateTest plan
python -m unittest tests/test_risco_cloud.py— all 18 tests passMaxRetriesErroris raised after 5 consecutive failed connection attempts🤖 Generated with Claude Code