Consume Subscriptions API

Create a subscription worker using get_subscription_worker or get_subscription_worker_by_name.

• Use the get_subscription_worker method to specify the subscription options while creating the worker.
• Use the get_subscription_worker_by_name method to create the worker using the default options.

def get_subscription_worker(
    self, options: SubscriptionWorkerOptions, object_type: Optional[Type[_T]] = None, database: Optional[str] = None
) -> SubscriptionWorker[_T]: ...

def get_subscription_worker_by_name(
    self,
    subscription_name: Optional[str] = None,
    object_type: Optional[Type[_T]] = None,
    database: Optional[str] = None,
) -> SubscriptionWorker[_T]: ...

When creating a worker with SubscriptionWorkerOptions, the only mandatory property is subscription_name.
All other parameters are optional and will default to their respective default values if not specified.

After creating a subscription worker, the subscription worker is still not processing any documents.
To start processing, you need to call the run function of the SubscriptionWorker.

The run function receives the client-side code as a function that will process the received document batches.

def run(self, process_documents: Optional[Callable[[SubscriptionBatch[_T]], Any]]) -> Future: ...

def number_of_items_in_batch(self) -> int:
    return 0 if self.items is None else len(self.items)

class Item(Generic[_T_Item]):
    """
    Represents a single item in a subscription batch results.
    This class should be used only inside the subscription's run delegate,
    using it outside this scope might cause unexpected behavior.
    """

class SubscriptionBatch(Generic[_T]):

def __init__(self):
    self._result: Optional[_T_Item] = None
    self._exception_message: Optional[str] = None
    self._key: Optional[str] = None
    self._change_vector: Optional[str] = None
    self._projection: Optional[bool] = None
    self._revision: Optional[bool] = None
    self.raw_result: Optional[Dict] = None
    self.raw_metadata: Optional[Dict] = None
    self._metadata: Optional[MetadataAsDictionary] = None

Methods:

Events:

Properties: