pyqt_reactive.services.interval_snapshot_poller

Generic interval-based background snapshot poller.

Classes

`CallbackIntervalSnapshotPollerPolicy`(...[, ...])	Callback-backed interval polling policy.
`IntervalSnapshotPoller`(policy)	Runs at most one background poll at a configured interval.
`IntervalSnapshotPollerPolicyABC`()	Policy boundary for background polling behavior.

class pyqt_reactive.services.interval_snapshot_poller.IntervalSnapshotPollerPolicyABC[source]

Policy boundary for background polling behavior.

abstractmethod fetch_snapshot() → TSnapshot[source]: Fetch latest snapshot from source.

abstractmethod clone_snapshot(snapshot: TSnapshot) → TSnapshot[source]: Return an isolated clone of one snapshot value.

property poll_interval_seconds: float

on_snapshot_changed(snapshot: TSnapshot) → None[source]

on_poll_error(error: Exception) → None[source]

class pyqt_reactive.services.interval_snapshot_poller.CallbackIntervalSnapshotPollerPolicy(fetch_snapshot_fn: Callable[[], TSnapshot], clone_snapshot_fn: Callable[[TSnapshot], TSnapshot], poll_interval_seconds_value: float = 1.0, on_snapshot_changed_fn: Callable[[TSnapshot], None] | None = None, on_poll_error_fn: Callable[[Exception], None] | None = None)[source]

Callback-backed interval polling policy.

fetch_snapshot_fn: Callable[[], TSnapshot]

clone_snapshot_fn: Callable[[TSnapshot], TSnapshot]

poll_interval_seconds_value: float = 1.0

on_snapshot_changed_fn: Callable[[TSnapshot], None] | None = None

on_poll_error_fn: Callable[[Exception], None] | None = None

fetch_snapshot() → TSnapshot[source]: Fetch latest snapshot from source.

clone_snapshot(snapshot: TSnapshot) → TSnapshot[source]: Return an isolated clone of one snapshot value.

property poll_interval_seconds: float

on_snapshot_changed(snapshot: TSnapshot) → None[source]

on_poll_error(error: Exception) → None[source]

__init__(fetch_snapshot_fn: Callable[[], TSnapshot], clone_snapshot_fn: Callable[[TSnapshot], TSnapshot], poll_interval_seconds_value: float = 1.0, on_snapshot_changed_fn: Callable[[TSnapshot], None] | None = None, on_poll_error_fn: Callable[[Exception], None] | None = None) → None

class pyqt_reactive.services.interval_snapshot_poller.IntervalSnapshotPoller(policy: IntervalSnapshotPollerPolicyABC[TSnapshot])[source]

Runs at most one background poll at a configured interval.

__init__(policy: IntervalSnapshotPollerPolicyABC[TSnapshot]) → None[source]

tick() → None[source]: Schedule one background poll if interval and inflight gates allow it.

get_snapshot_copy() → TSnapshot | None[source]: Read the current snapshot using policy clone semantics.

is_poll_inflight() → bool[source]

reset() → None[source]: Drop current snapshot and invalidate any in-flight worker result.