# fiftyone.operators.store.notification_service Notification service for ExecutionStore using MongoDB Change Streams. Copyright 2017-2026, Voxel51, Inc.
[voxel51.com](https://voxel51.com/)

**Classes:** | [`ChangeStreamNotificationService`](#fiftyone.operators.store.notification_service.ChangeStreamNotificationService)() | Abstract base class for change stream notification services. | |--------------------------------------------------------------------------------------------------------------------------------------------------------------------|----------------------------------------------------------------| | [`MongoChangeStreamNotificationService`](#fiftyone.operators.store.notification_service.MongoChangeStreamNotificationService)(...[, ...]) | | | [`MongoChangeStreamNotificationServiceLifecycleManager`](#fiftyone.operators.store.notification_service.MongoChangeStreamNotificationServiceLifecycleManager)(...) | | **Functions:** | [`is_notification_service_disabled`](#fiftyone.operators.store.notification_service.is_notification_service_disabled)() | Check if the notification service is disabled. | |---------------------------------------------------------------------------------------------------------------------------|--------------------------------------------------| ### *class* fiftyone.operators.store.notification_service.ChangeStreamNotificationService Bases: `ABC` Abstract base class for change stream notification services. **Methods:** | [`subscribe`](#fiftyone.operators.store.notification_service.ChangeStreamNotificationService.subscribe)(store_name, callback[, dataset_id]) | Register a local subscriber for a specific store. | |-----------------------------------------------------------------------------------------------------------------------------------------------|------------------------------------------------------------| | [`unsubscribe`](#fiftyone.operators.store.notification_service.ChangeStreamNotificationService.unsubscribe)(subscription_id) | Unsubscribe local subscribers from a specific store. | | [`unsubscribe_all`](#fiftyone.operators.store.notification_service.ChangeStreamNotificationService.unsubscribe_all)(store_name) | Unsubscribe from all changes in a store. | | [`notify`](#fiftyone.operators.store.notification_service.ChangeStreamNotificationService.notify)(store_name, message_data) | Notify local subscribers and remote listeners of a change. | | [`start`](#fiftyone.operators.store.notification_service.ChangeStreamNotificationService.start)() | Start watching for database changes. | | [`stop`](#fiftyone.operators.store.notification_service.ChangeStreamNotificationService.stop)() | Stop watching for database changes. | #### *abstractmethod* subscribe(store_name: str, callback: Callable[[str], None], dataset_id: str | None = None) → str Register a local subscriber for a specific store. * **Parameters:** * **store_name** – The name of the store to subscribe to. * **callback** – The callback to call when a change occurs. * **dataset_id** – Optional dataset ID to filter changes by. * **Returns:** The subscription id. #### *abstractmethod* unsubscribe(subscription_id: str) Unsubscribe local subscribers from a specific store. * **Parameters:** **subscription_id** – The subscription id to unsubscribe from. #### *abstractmethod* unsubscribe_all(store_name: str) Unsubscribe from all changes in a store. * **Parameters:** **store_name** (*str*) – the name of the store to unsubscribe from #### *abstractmethod* notify(store_name: str, message_data: [MessageData](fiftyone.operators.message.md#fiftyone.operators.message.MessageData)) → None Notify local subscribers and remote listeners of a change. * **Parameters:** * **store_name** – The name of the store that changed. * **message** – The message to notify subscribers with. #### *abstractmethod async* start() → None Start watching for database changes. #### *abstractmethod async* stop() → None Stop watching for database changes. ### *class* fiftyone.operators.store.notification_service.MongoChangeStreamNotificationService(collection_name: str, remote_notifier: [RemoteNotifier](fiftyone.operators.remote_notifier.md#fiftyone.operators.remote_notifier.RemoteNotifier) = None, registry: [LocalSubscriptionRegistry](fiftyone.operators.store.subscription_registry.md#fiftyone.operators.store.subscription_registry.LocalSubscriptionRegistry) = None) Bases: [`ChangeStreamNotificationService`](#fiftyone.operators.store.notification_service.ChangeStreamNotificationService) **Methods:** | [`subscribe`](#fiftyone.operators.store.notification_service.MongoChangeStreamNotificationService.subscribe)(store_name, callback[, dataset_id]) | Register a local subscriber for a specific store. | |----------------------------------------------------------------------------------------------------------------------------------------------------|----------------------------------------------------------------------------| | [`unsubscribe`](#fiftyone.operators.store.notification_service.MongoChangeStreamNotificationService.unsubscribe)(subscription_id) | Unsubscribe from a specific store. | | [`unsubscribe_all`](#fiftyone.operators.store.notification_service.MongoChangeStreamNotificationService.unsubscribe_all)(store_name) | Unsubscribe from all changes in a store. | | [`start`](#fiftyone.operators.store.notification_service.MongoChangeStreamNotificationService.start)(dedicated_event_loop) | Start watching the collection for changes using change streams or polling. | | [`notify`](#fiftyone.operators.store.notification_service.MongoChangeStreamNotificationService.notify)(store_name, message_data) | Notify local subscribers and remote listeners of a change. | | [`stop`](#fiftyone.operators.store.notification_service.MongoChangeStreamNotificationService.stop)() | Signal stop watching the collection for changes. | #### subscribe(store_name: str, callback: Callable[[str], None], dataset_id: str | None = None) → str Register a local subscriber for a specific store. * **Parameters:** * **store_name** – The name of the store to subscribe to. * **callback** – The callback to call when a change occurs. * **dataset_id** – Optional dataset ID to filter changes by. * **Returns:** The subscription id. #### unsubscribe(subscription_id: str) Unsubscribe from a specific store. * **Parameters:** **subscription_id** – The subscription id to unsubscribe from. #### unsubscribe_all(store_name: str) Unsubscribe from all changes in a store. * **Parameters:** **store_name** (*str*) – the name of the store to unsubscribe from #### *async* start(dedicated_event_loop: AbstractEventLoop) → None Start watching the collection for changes using change streams or polling. #### *async* notify(store_name: str, message_data: [MessageData](fiftyone.operators.message.md#fiftyone.operators.message.MessageData)) → None Notify local subscribers and remote listeners of a change. Handles exceptions gracefully to prevent failures when clients disconnect. * **Parameters:** * **store_name** – The name of the store that changed * **message_data** – The message data to notify subscribers with #### *async* stop() → None Signal stop watching the collection for changes. Assume this is called from thread safe context. ### *class* fiftyone.operators.store.notification_service.MongoChangeStreamNotificationServiceLifecycleManager(notification_service: [MongoChangeStreamNotificationService](#fiftyone.operators.store.notification_service.MongoChangeStreamNotificationService)) Bases: `object` **Methods:** | [`start_in_dedicated_thread`](#fiftyone.operators.store.notification_service.MongoChangeStreamNotificationServiceLifecycleManager.start_in_dedicated_thread)() | Create a dedicated event loop in a new thread and start the notification service. | |------------------------------------------------------------------------------------------------------------------------------------------------------------------|-------------------------------------------------------------------------------------| | [`stop`](#fiftyone.operators.store.notification_service.MongoChangeStreamNotificationServiceLifecycleManager.stop)() | | #### start_in_dedicated_thread() → None Create a dedicated event loop in a new thread and start the notification service. #### *async* stop() → None ### fiftyone.operators.store.notification_service.is_notification_service_disabled() → [bool](fiftyone.core.stages.md#fiftyone.core.stages.Exists.bool) Check if the notification service is disabled.