Bytewax Valkey

Valkey connectors for Bytewax.

This connector offers 3 sources and 2 sinks:

• StreamSink - writes Valkey streams using xadd
• StreamSource - reads Valkey streams using xread
• PubSubSink - writes Valkey pubsub using publish
• PubSubSource - reads Valkey pubsub using subscribe
• PubSubPatternSource - reads Valkey pubsub using psubscribe

Installation

This package is available via PyPi as bytewax-valkey and can be installed via your package manager of choice.

Usage

Pub/Sub Source

import os

from bytewax_valkey import PubSubSource
from bytewax.connectors.stdio import StdOutSink

import bytewax.operators as op
from bytewax.dataflow import Dataflow

VALKEY_URL = os.environ["VALKEY_URL"]

flow = Dataflow("valkey_example")
flow_input = op.input("input", flow, PubSubSource.from_url(VALKEY_URL, "example"))
op.output("output", flow_input, StdOutSink())

Pub/Sub Pattern Source

import os

from bytewax_valkey import PubSubPatternSource
from bytewax.connectors.stdio import StdOutSink

import bytewax.operators as op
from bytewax.dataflow import Dataflow

VALKEY_URL = os.environ["VALKEY_URL"]

flow = Dataflow("valkey_example")
flow_input = op.input("input", flow, PubSubPatternSource.from_url(VALKEY_URL, "example*"))
op.output("output", flow_input, StdOutSink())

Pub/Sub Sink

import os

from bytewax_valkey import PubSubSink
from bytewax.testing import TestingSource

import bytewax.operators as op
from bytewax.dataflow import Dataflow

VALKEY_URL = os.environ["VALKEY_URL"]

flow = Dataflow("valkey_example")
flow_input = op.input("input", flow, TestingSource([b"example message"]))
op.output("output", flow_input, PubSubSink.from_url(VALKEY_URL, "example"))

Stream Source

import os

from bytewax_valkey import StreamSource
from bytewax.connectors.stdio import StdOutSink

import bytewax.operators as op
from bytewax.dataflow import Dataflow

VALKEY_URL = os.environ["VALKEY_URL"]

flow = Dataflow("valkey_example")
flow_input = op.input("input", flow, StreamSource.from_url(VALKEY_URL, "example"))
op.output("output", flow_input, StdOutSink())

Stream Sink

import os

from bytewax_valkey import StreamSink
from bytewax.testing import TestingSource

import bytewax.operators as op
from bytewax.dataflow import Dataflow

VALKEY_URL = os.environ["VALKEY_URL"]

flow = Dataflow("valkey_example")
flow_input = op.input("input", flow, TestingSource([{"key": "value"}]))
op.output("output", flow_input, StreamSink.from_url(VALKEY_URL, "example"))

License

Licensed under the MIT License.