Skip to main content

Python client for Liftbridge.

Project description

python-liftbridge

PyPI PyPI - License

Python client for Liftbridge, a system that provides lightweight, fault-tolerant message streams for NATS.

Liftbridge provides the following high-level features:

  • Log-based API for NATS
  • Replicated for fault-tolerance
  • Horizontally scalable
  • Wildcard subscription support
  • At-least-once delivery support and message replay
  • Message key-value support
  • Log compaction by key

Installation

$ pip install python-liftbridge

Basic Usage

from python_liftbridge import Lift, Message, Stream, ErrStreamExists

# Create a Liftbridge client.
client = Lift(ip_address='localhost:9292', timeout=5)

# Create a stream attached to the NATS subject "foo".
try:
    client.create_stream(Stream(subject='foo', name='foo-stream'))
except ErrStreamExists:
    print('This stream already exists!')

# Publish a message to "foo".
client.publish(Message(value='hello', subject='foo'))

# Subscribe to the stream starting from the beginning.
for message in client.subscribe(
    Stream(
        subject='foo',
        name='foo-stream',
    ).start_at_earliest_received(),
):
    print("Received: '{}'".format(message.value))

Create Stream

Streams are a durable message log attached to a NATS subject. They record messages published to the subject for consumption.

Streams have a few key properties: a subject, which is the corresponding NATS subject, a name, which is a human-readable identifier for the stream, and a replication factor, which is the number of nodes the stream should be replicated to for redundancy. Optionally, there is a group which is the name of a load-balance group for the stream to join. When there are multiple streams in the same group, messages will be balanced among them.

"""
    Create a stream attached to the NATS subject "foo.*" that is replicated to
    all the brokers in the cluster. ErrStreamExists is returned if a stream with
    the given name already exists for the subject.
"""
client.create_stream(Stream(subject='foo.*', name='my-stream', max_replication=True))

Subscription Start/Replay Options

Subscriptions are how Liftbridge streams are consumed. Clients can choose where to start consuming messages from in a stream. This is controlled using options passed to Subscribe.

# Subscribe starting with new messages only.
client.subscribe(
    Stream(subject='foo', name='foo-stream')
)
# Subscribe starting with the most recently published value.
client.subscribe(
    Stream(subject='foo', name='foo-stream').start_at_earliest_received()
)
# Subscribe starting with the oldest published value.
client.subscribe(
    Stream(subject='foo', name='foo-stream').start_at_latest_received()
)
# Subscribe starting at a specific offset.
client.subscribe(
    Stream(subject='foo', name='foo-stream').start_at_offset(4)
)
# Subscribe starting at a specific time.
client.subscribe(
    Stream(subject='foo', name='foo-stream').start_at_time(datetime.now())
)
# Subscribe starting at a specific amount of time in the past.
client.subscribe(
    Stream(subject='foo', name='foo-stream').start_at_time_delta(timedelta(days=1))
)

Publishing

A publish API is provided to make it easy to write messages to streams. This includes a number of options for decorating messages with metadata like a message key.

Keys are used by Liftbridge's log compaction. When enabled, Liftbridge streams will retain only the last message for a given key.

# Publish a message with a key
client.publish(Message(subject='foo', value='Hello', key='key'))

Publishing Directly with NATS

Since Liftbridge is an extension of NATS, a NATS client can also be used to publish messages. This means existing NATS publishers do not need any changes for messages to be consumed in Liftbridge.

How to contribute

  1. Check for open issues or open a fresh issue to start a discussion around a feature idea or a bug.
  2. Fork the repository on GitHub to start making your changes to the master branch (or branch off of it).
  3. Write a test which shows that the bug was fixed or that the feature works as expected.
  4. Send a pull request and bug me until it gets merged and published.

Some things on the backlog:

  • <input type="checkbox" disabled="" /> Add documentation (Sphynx)
  • <input type="checkbox" disabled="" /> Add CI (CircleCI)
  • <input type="checkbox" disabled="" /> Add tests
  • <input type="checkbox" disabled="" /> Add code coverage
  • <input type="checkbox" disabled="" /> Add TLS support for gRPC
  • <input type="checkbox" disabled="" /> Add message headers support
  • <input type="checkbox" disabled="" /> Add message ACK support (scaffolding is already done)
  • <input type="checkbox" disabled="" /> Add method to close connection
  • <input type="checkbox" disabled="" /> Add async client
  • <input type="checkbox" disabled="" /> Add gRPC connection pool
  • <input type="checkbox" disabled="" /> Improve fetch metadata
  • <input type="checkbox" disabled="" /> Better instrumentation/observability (OpenCensus support?)

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Files for python-liftbridge, version 0.0.1
Filename, size File type Python version Upload date Hashes
Filename, size python_liftbridge-0.0.1-py3-none-any.whl (7.5 kB) File type Wheel Python version py3 Upload date Hashes View
Filename, size python-liftbridge-0.0.1.tar.gz (3.7 kB) File type Source Python version None Upload date Hashes View

Supported by

AWS AWS Cloud computing Datadog Datadog Monitoring Facebook / Instagram Facebook / Instagram PSF Sponsor Fastly Fastly CDN Google Google Object Storage and Download Analytics Huawei Huawei PSF Sponsor Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Salesforce Salesforce PSF Sponsor Sentry Sentry Error logging StatusPage StatusPage Status page