fluvio

  1from ._fluvio_python import (
  2    Fluvio as _Fluvio,
  3    ConsumerConfig as _ConsumerConfig,
  4    PartitionConsumer as _PartitionConsumer,
  5    PartitionConsumerStream as _PartitionConsumerStream,
  6    TopicProducer as _TopicProducer,
  7    ProducerBatchRecord as _ProducerBatchRecord,
  8    SmartModuleKind as _SmartModuleKind,
  9    Record as _Record,
 10    Offset as _Offset,
 11)
 12from enum import Enum
 13from ._fluvio_python import Error as FluviorError  # noqa: F401
 14import typing
 15
 16
 17class Record:
 18    """The individual record for a given stream."""
 19
 20    _inner: _Record
 21
 22    def __init__(self, inner: _Record):
 23        self._inner = inner
 24
 25    def offset(self) -> int:
 26        """The offset from the initial offset for a given stream."""
 27        return self._inner.offset()
 28
 29    def value(self) -> typing.List[int]:
 30        """Returns the contents of this Record's value"""
 31        return self._inner.value()
 32
 33    def value_string(self) -> str:
 34        """The UTF-8 decoded value for this record."""
 35        return self._inner.value_string()
 36
 37    def key(self) -> typing.List[int]:
 38        """Returns the contents of this Record's key, if it exists"""
 39        return self._inner.key()
 40
 41    def key_string(self) -> str:
 42        """The UTF-8 decoded key for this record."""
 43        return self._inner.key_string()
 44
 45
 46class Offset:
 47    """Describes the location of an event stored in a Fluvio partition."""
 48
 49    _inner: _Offset
 50
 51    @classmethod
 52    def absolute(cls, index: int):
 53        """Creates an absolute offset with the given index"""
 54        return cls(_Offset.absolute(index))
 55
 56    @classmethod
 57    def beginning(cls):
 58        """Creates a relative offset starting at the beginning of the saved log"""
 59        return cls(_Offset.beginning())
 60
 61    @classmethod
 62    def end(cls):
 63        """Creates a relative offset pointing to the newest log entry"""
 64        return cls(_Offset.end())
 65
 66    @classmethod
 67    def from_beginning(cls, offset: int):
 68        """Creates a relative offset a fixed distance after the oldest log
 69        entry
 70        """
 71        return cls(_Offset.from_beginning(offset))
 72
 73    @classmethod
 74    def from_end(cls, offset: int):
 75        """Creates a relative offset a fixed distance before the newest log
 76        entry
 77        """
 78        return cls(_Offset.from_end(offset))
 79
 80    def __init__(self, inner: _Offset):
 81        self._inner = inner
 82
 83
 84class SmartModuleKind(Enum):
 85    """
 86    Use of this is to explicitly set the kind of a smartmodule. Not required
 87    but needed for legacy SmartModules.
 88    """
 89
 90    Filter = _SmartModuleKind.Filter
 91    Map = _SmartModuleKind.Map
 92    ArrayMap = _SmartModuleKind.ArrayMap
 93    FilterMap = _SmartModuleKind.FilterMap
 94    Aggregate = _SmartModuleKind.Aggregate
 95
 96
 97class ConsumerConfig:
 98    _inner: _ConsumerConfig
 99
100    def __init__(self):
101        self._inner = _ConsumerConfig()
102
103    def smartmodule(
104        self,
105        name: str = None,
106        path: str = None,
107        kind: SmartModuleKind = None,
108        params: typing.Dict[str, str] = None,
109        aggregate: typing.List[bytes] = None,
110    ):
111        """
112        This is a method for adding a smartmodule to a consumer config either
113        using a `name` of a `SmartModule` or a `path` to a wasm binary.
114
115        Args:
116
117            name: str
118            path: str
119            kind: SmartModuleKind
120            params: Dict[str, str]
121            aggregate: List[bytes] # This is used for the initial value of an aggregate smartmodule
122
123        Raises:
124            "Require either a path or a name for a smartmodule."
125            "Only specify one of path or name not both."
126
127        Returns:
128
129            None
130        """
131
132        if kind is not None:
133            kind = kind.value
134
135        if path is None and name is None:
136            raise Exception("Require either a path or a name for a smartmodule.")
137
138        if path is not None and name is not None:
139            raise Exception("Only specify one of path or name not both.")
140
141        params = {} if params is None else params
142        param_keys = [x for x in params.keys()]
143        param_vals = [x for x in params.values()]
144
145        self._inner.smartmodule(
146            name,
147            path,
148            kind,
149            param_keys,
150            param_vals,
151            aggregate,
152            # These arguments are for Join stuff but that's not implemented on
153            # the python side yet
154            None,
155            None,
156            None,
157            None,
158        )
159
160
161class PartitionConsumer:
162    """
163    An interface for consuming events from a particular partition
164
165    There are two ways to consume events: by "fetching" events and by
166    "streaming" events. Fetching involves specifying a range of events that you
167    want to consume via their Offset. A fetch is a sort of one-time batch
168    operation: you’ll receive all of the events in your range all at once. When
169    you consume events via Streaming, you specify a starting Offset and receive
170    an object that will continuously yield new events as they arrive.
171    """
172
173    _inner: _PartitionConsumer
174
175    def __init__(self, inner: _PartitionConsumer):
176        self._inner = inner
177
178    def stream(self, offset: Offset) -> typing.Iterator[Record]:
179        """
180        Continuously streams events from a particular offset in the consumer’s
181        partition. This returns a `Iterator[Record]` which is an
182        iterator.
183
184        Streaming is one of the two ways to consume events in Fluvio. It is a
185        continuous request for new records arriving in a partition, beginning
186        at a particular offset. You specify the starting point of the stream
187        using an Offset and periodically receive events, either individually or
188        in batches.
189        """
190        return self._generator(self._inner.stream(offset._inner))
191
192    def stream_with_config(
193        self, offset: Offset, config: ConsumerConfig
194    ) -> typing.Iterator[Record]:
195        """
196        Continuously streams events from a particular offset with a SmartModule
197        WASM module in the consumer’s partition. This returns a
198        `Iterator[Record]` which is an iterator.
199
200        Streaming is one of the two ways to consume events in Fluvio. It is a
201        continuous request for new records arriving in a partition, beginning
202        at a particular offset. You specify the starting point of the stream
203        using an Offset and periodically receive events, either individually or
204        in batches.
205
206        Args:
207            offset: Offset
208            wasm_module_path: str - The absolute path to the WASM file
209
210        Example:
211            import os
212
213            wmp = os.path.abspath("somefilter.wasm")
214            config = ConsumerConfig()
215            config.smartmodule(path=wmp)
216            for i in consumer.stream_with_config(Offset.beginning(), config):
217                # do something with i
218
219        Returns:
220            `Iterator[Record]`
221
222        """
223        return self._generator(
224            self._inner.stream_with_config(offset._inner, config._inner)
225        )
226
227    def _generator(self, stream: _PartitionConsumerStream) -> typing.Iterator[Record]:
228        item = stream.next()
229        while item is not None:
230            yield Record(item)
231            item = stream.next()
232
233
234class TopicProducer:
235    """An interface for producing events to a particular topic.
236
237    A `TopicProducer` allows you to send events to the specific topic it was
238    initialized for. Once you have a `TopicProducer`, you can send events to
239    the topic, choosing which partition each event should be delivered to.
240    """
241
242    _inner: _TopicProducer
243
244    def __init__(self, inner: _TopicProducer):
245        self._inner = inner
246
247    def send_string(self, buf: str) -> None:
248        """Sends a string to this producer’s topic"""
249        return self.send([], buf.encode("utf-8"))
250
251    def send(self, key: typing.List[int], value: typing.List[int]) -> None:
252        """
253        Sends a key/value record to this producer's Topic.
254
255        The partition that the record will be sent to is derived from the Key.
256        """
257        return self._inner.send(key, value)
258
259    def flush(self) -> None:
260        """
261        Send all the queued records in the producer batches.
262        """
263        return self._inner.flush()
264
265    def send_all(self, records: typing.List[typing.Tuple[bytes, bytes]]):
266        """
267        Sends a list of key/value records as a batch to this producer's Topic.
268        :param records: The list of records to send
269        """
270        records_inner = [_ProducerBatchRecord(x, y) for (x, y) in records]
271        return self._inner.send_all(records_inner)
272
273
274class Fluvio:
275    """An interface for interacting with Fluvio streaming."""
276
277    _inner: _Fluvio
278
279    def __init__(self, inner: _Fluvio):
280        self._inner = inner
281
282    @classmethod
283    def connect(cls):
284        """Creates a new Fluvio client using the current profile from
285        `~/.fluvio/config`
286
287        If there is no current profile or the `~/.fluvio/config` file does not
288        exist, then this will create a new profile with default settings and
289        set it as current, then try to connect to the cluster using those
290        settings.
291        """
292        return cls(_Fluvio.connect())
293
294    def partition_consumer(self, topic: str, partition: int) -> PartitionConsumer:
295        """Creates a new `PartitionConsumer` for the given topic and partition
296
297        Currently, consumers are scoped to both a specific Fluvio topic and to
298        a particular partition within that topic. That means that if you have a
299        topic with multiple partitions, then in order to receive all of the
300        events in all of the partitions, you will need to create one consumer
301        per partition.
302        """
303        return PartitionConsumer(self._inner.partition_consumer(topic, partition))
304
305    def topic_producer(self, topic: str) -> TopicProducer:
306        """
307        Creates a new `TopicProducer` for the given topic name.
308
309        Currently, producers are scoped to a specific Fluvio topic. That means
310        when you send events via a producer, you must specify which partition
311        each event should go to.
312        """
313        return TopicProducer(self._inner.topic_producer(topic))