simple

Implements SimpleConsumer, a file-based consumer for local development and testing.

class unistream.consumers.simple.SimpleConsumer(record_class: type[AbcRecord] = REQ, limit: int = REQ, checkpoint: BaseCheckPoint = REQ, exp_backoff_multiplier: int = REQ, exp_backoff_base: int = REQ, exp_backoff_min: int = REQ, exp_backoff_max: int = REQ, skip_error: bool = REQ, delay: int | float = REQ, path_source: Path = REQ, path_dlq: Path = REQ)[source]

A simple consumer that read data from a local, append-only log file.

It is a good example to show how to implement a consumer and understand the behavior of the consumer.

You should use it along with SimpleProducer.

Note

Don’t initialize this class directly, use the SimpleConsumer.new() method

Parameters:

  • record_class – the record class.

  • limit – the max number of records to fetch from the stream system.

  • checkpoint – the BaseCheckpoint object for status tracking and fault tolerance.

  • exp_backoff_multiplier – the multiplier of the exponential backoff

  • exp_backoff_base – the base of the exponential backoff

  • exp_backoff_min – the minimum wait time of the exponential backoff

  • exp_backoff_max – the maximum wait time of the exponential backoff

  • skip_error – if True, skip the error and continue to process the next record. this is the most common use case. if False, raise the error and stop the consumer.

  • delay – the delay time between pulling two batches.

  • path_source – the path of the source file to read from.

  • path_dlq – the path of the dead letter queue file to write to.

classmethod new(record_class: type[AbcRecord], path_source: Path, path_dlq: Path, checkpoint: BaseCheckPoint, limit: int = 1000, exp_backoff_multiplier: int = 1, exp_backoff_base: int = 2, exp_backoff_min: int = 1, exp_backoff_max: int = 60, skip_error: bool = True, delay: int | float = 0)[source]

Factory method to create a consumer.

get_records(limit: int | None = None) → tuple[list[AbcRecord], str | int][source]

Get records from the stream system and determine the value of the next pointer for the next batch if we successfully process this batch of records.

Parameters:

limit – The maximum number of records to return.

Returns:

a two-item tuple. the first one is a list of records and the second one is the value of the next pointer for the next batch if we successfully process this batch of records.

Important

User has to implement this method.

If you need additional parameters other than the BaseConsumer built-in attributes and limit, you should extend this class and add the parameters to the subclass.