(
        self,
        api_name,
        event_name,
        start: datetime = None,
        stop: datetime = None,
        start_inclusive: bool = True,
        batch_size: int = 100,
    )
Retrieve historical events for the given API

        Will not have any impact on existing consumer groups.
Retrieve historical events for the given API
Retrieve historical events for the given API
async def history(
        self,
        api_name,
        event_name,
        start: datetime = None,
        stop: datetime = None,
        start_inclusive: bool = True,
        batch_size: int = 100,
    ) -> AsyncGenerator[EventMessage, None]:
        """Retrieve historical events for the given API

        Will not have any impact on existing consumer groups.
        """
        redis_start = datetime_to_redis_steam_id(start) if start else "-"
        redis_stop = datetime_to_redis_steam_id(stop) if stop else "+"

        if start and not start_inclusive:
            redis_start = redis_stream_id_add_one(redis_start)

        stream_name = self._get_stream_names([(api_name, event_name)])[0]

        logger.debug(
            f"Getting history for stream {stream_name} from {redis_start} ({start}) "
            f"to {redis_stop} ({stop}) in batches of {batch_size}"
        )

        with await self.connection_manager() as redis:
            messages = True
            while messages:
                messages = await redis.xrevrange(
                    stream_name, redis_stop, redis_start, count=batch_size
                )
                if not messages:
                    return
                for message_id, fields in messages:
                    message_id = decode(message_id, "utf8")
                    redis_stop = redis_stream_id_subtract_one(message_id)
                    try:
                        event_message = self._fields_to_message(
                            fields,
                            expected_event_names={event_name},
                            stream=stream_name,
                            native_id=message_id,
                            consumer_group=None,
                        )
                    except (NoopMessage, IgnoreMessage):
                        logger.debug(
                            f"Ignoring NOOP event with ID {message_id} discovered during fetching"
                            " of event history"
                        )
                    else:
                        yield event_message
async def history ( self , api_name , event_name , start : datetime = None , stop : datetime = None , start_inclusive : bool = True , batch_size : int = 100 , ) -> AsyncGenerator [ EventMessage , None ] : redis_start = datetime_to_redis_steam_id ( start ) if start else "-" redis_stop = datetime_to_redis_steam_id ( stop ) if stop else "+" if start and not start_inclusive : redis_start = redis_stream_id_add_one ( redis_start ) stream_name = self . _get_stream_names ( [ ( api_name , event_name ) ] ) [ 0 ] logger . debug ( f"Getting history for stream {stream_name} from {redis_start} ({start}) " f"to {redis_stop} ({stop}) in batches of {batch_size}" ) with await self . connection_manager ( ) as redis : messages = True while messages : messages = await redis . xrevrange ( stream_name , redis_stop , redis_start , count = batch_size ) if not messages : return for message_id , fields in messages : message_id = decode ( message_id , "utf8" ) redis_stop = redis_stream_id_subtract_one ( message_id ) try : event_message = self . _fields_to_message ( fields , expected_event_names = { event_name } , stream = stream_name , native_id = message_id , consumer_group = None , ) except ( NoopMessage , IgnoreMessage ) : logger . debug ( f"Ignoring NOOP event with ID {message_id} discovered during fetching" " of event history" ) else : yield event_message