flo-python

Python SDK for the Flo distributed systems platform.

Installation

pip install flo

Quick Start

import asyncio
from flo import FloClient

async def main():
    # Connect to Flo server
    async with FloClient("localhost:9000") as client:
        # KV operations
        await client.kv.put("user:123", b"John Doe")
        value = await client.kv.get("user:123")
        print(f"Got: {value.decode()}")

        # Queue operations
        seq = await client.queue.enqueue("tasks", b'{"task": "process"}')
        print(f"Enqueued message with seq: {seq}")

asyncio.run(main())

Features

KV Store: Versioned key-value storage with MVCC, TTL, and optimistic locking
Queues: Priority-based task queues with visibility timeout and dead letter queues
Streams: Append-only logs with consumer groups for distributed processing
Actions: Registered tasks with configurable timeouts, retries, and idempotency
Workers: Distributed task execution with lease management and heartbeats
Async/await: Native asyncio support for high-performance applications
Type hints: Full type annotations for IDE support

KV Operations

Get

from flo import GetOptions

# Simple get
value = await client.kv.get("key")
if value is not None:
    print(value.decode())

# Blocking get - wait up to 5 seconds for value to appear
value = await client.kv.get("key", GetOptions(block_ms=5000))

# Blocking get - wait indefinitely (0 = infinite)
value = await client.kv.get("key", GetOptions(block_ms=0))

# Get with namespace override
value = await client.kv.get("key", GetOptions(namespace="other"))

Put

from flo import PutOptions

# Simple put
await client.kv.put("key", b"value")

# Put with TTL (expires in 1 hour)
await client.kv.put("session:abc", b"data", PutOptions(ttl_seconds=3600))

# Put with CAS (optimistic locking)
await client.kv.put("counter", b"2", PutOptions(cas_version=1))

# Put only if key doesn't exist
await client.kv.put("key", b"value", PutOptions(if_not_exists=True))

# Put only if key exists
await client.kv.put("key", b"value", PutOptions(if_exists=True))

Delete

await client.kv.delete("key")

Scan

from flo import ScanOptions

# Scan all keys with prefix
result = await client.kv.scan("user:")
for entry in result.entries:
    print(f"{entry.key.decode()}: {entry.value.decode()}")

# Paginated scan
result = await client.kv.scan("user:", ScanOptions(limit=100))
while result.has_more:
    # Process entries...
    result = await client.kv.scan("user:", ScanOptions(cursor=result.cursor, limit=100))

# Keys only (more efficient when you don't need values)
result = await client.kv.scan("user:", ScanOptions(keys_only=True))

History

from flo import HistoryOptions

# Get version history for a key
history = await client.kv.history("user:123", HistoryOptions(limit=10))
for entry in history:
    print(f"v{entry.version} at {entry.timestamp}: {entry.value.decode()}")

Queue Operations

Enqueue

from flo import EnqueueOptions

# Simple enqueue
seq = await client.queue.enqueue("tasks", b'{"task": "process"}')

# With priority (higher = more urgent, 0-255)
seq = await client.queue.enqueue("tasks", payload, EnqueueOptions(priority=10))

# With delay (message invisible for 60 seconds)
seq = await client.queue.enqueue("tasks", payload, EnqueueOptions(delay_ms=60000))

# With deduplication key
seq = await client.queue.enqueue("tasks", payload, EnqueueOptions(dedup_key="task-123"))

Dequeue

from flo import DequeueOptions

# Dequeue up to 10 messages
result = await client.queue.dequeue("tasks", 10)
for msg in result.messages:
    print(f"Processing message {msg.seq}: {msg.payload}")

# Long polling (wait up to 30s for messages)
result = await client.queue.dequeue("tasks", 10, DequeueOptions(block_ms=30000))

# Custom visibility timeout (message invisible for 60s)
result = await client.queue.dequeue("tasks", 10, DequeueOptions(visibility_timeout_ms=60000))

Ack/Nack

from flo import NackOptions

result = await client.queue.dequeue("tasks", 10)
for msg in result.messages:
    try:
        process(msg.payload)
        # Acknowledge successful processing
        await client.queue.ack("tasks", [msg.seq])
    except Exception:
        # Retry the message
        await client.queue.nack("tasks", [msg.seq])
        # Or send to DLQ
        await client.queue.nack("tasks", [msg.seq], NackOptions(to_dlq=True))

Dead Letter Queue

from flo import DlqListOptions

# List DLQ messages
result = await client.queue.dlq_list("tasks", DlqListOptions(limit=100))
for msg in result.messages:
    print(f"Failed message {msg.seq}: {msg.payload}")

# Requeue DLQ messages
seqs = [msg.seq for msg in result.messages]
await client.queue.dlq_requeue("tasks", seqs)

Peek

# Peek at messages without creating leases (no visibility timeout)
result = await client.queue.peek("tasks", 5)
for msg in result.messages:
    print(f"Message {msg.seq}: {msg.payload}")
# Messages remain visible to other consumers

Touch (Lease Renewal)

# Renew lease on messages during long-running processing
result = await client.queue.dequeue("tasks", 1)
msg = result.messages[0]

# Long running task - periodically touch to prevent visibility timeout
for chunk in process_in_chunks(msg.payload):
    await process_chunk(chunk)
    await client.queue.touch("tasks", [msg.seq])  # Renew lease

await client.queue.ack("tasks", [msg.seq])

Stream Operations

Streams are append-only logs ideal for event sourcing, activity feeds, and real-time data pipelines.

Append

from flo import StreamAppendOptions

# Simple append
result = await client.stream.append("events", b'{"event": "click", "user": "123"}')
print(f"Appended at offset {result.offset}, timestamp {result.timestamp}")

# Append with partition key (for ordered processing within partition)
result = await client.stream.append(
    "events", payload,
    StreamAppendOptions(partition_key="user-123")
)

# Append with client-provided timestamp
result = await client.stream.append(
    "events", payload,
    StreamAppendOptions(timestamp=1699999999000)
)

Read

from flo import StreamReadOptions, StreamStartMode

# Read from beginning (default)
result = await client.stream.read("events")
for record in result.records:
    print(f"offset={record.offset} ts={record.timestamp}: {record.payload}")

# Read from specific offset
result = await client.stream.read(
    "events",
    StreamReadOptions(start_mode=StreamStartMode.OFFSET, offset=100, count=10)
)

# Read from tail (latest records)
result = await client.stream.read(
    "events",
    StreamReadOptions(start_mode=StreamStartMode.TAIL, count=10)
)

# Read from timestamp
result = await client.stream.read(
    "events",
    StreamReadOptions(start_mode=StreamStartMode.TIMESTAMP, timestamp=1699999999000)
)

# Blocking read (long polling) - wait up to 30s for new records
result = await client.stream.read(
    "events",
    StreamReadOptions(offset=100, block_ms=30000)
)

Consumer Groups

Consumer groups allow multiple consumers to process a stream in parallel, with each record delivered to only one consumer.

from flo import StreamGroupReadOptions

# Join a consumer group
await client.stream.group_join("events", "processors", "worker-1")

# Read from consumer group
result = await client.stream.group_read("events", "processors", "worker-1")
for record in result.records:
    try:
        process(record.payload)
        # Acknowledge successful processing
        await client.stream.group_ack("events", "processors", [record.id])
    except Exception:
        # Record will be redelivered to another consumer
        pass

# With blocking (long polling)
result = await client.stream.group_read(
    "events", "processors", "worker-1",
    StreamGroupReadOptions(count=10, block_ms=30000)
)

Action Operations

Actions are registered tasks that can be invoked and executed by workers.

Register an Action

from flo import ActionRegisterOptions, ActionType

# Register a user-based action (executed by external workers)
await client.action.register(
    "process-image",
    ActionType.USER,
    ActionRegisterOptions(timeout_ms=60000, max_retries=3)
)

Invoke an Action

from flo import ActionInvokeOptions

# Invoke an action
result = await client.action.invoke(
    "process-image",
    b'{"image_url": "https://example.com/image.jpg"}',
    ActionInvokeOptions(priority=10)
)
print(f"Run ID: {result.run_id}")

# Invoke with idempotency key (prevents duplicate runs)
result = await client.action.invoke(
    "process-image",
    payload,
    ActionInvokeOptions(idempotency_key="order-123-image")
)

Check Action Status

status = await client.action.status(result.run_id)
print(f"Status: {status.status}")
if status.result:
    print(f"Result: {status.result}")

List and Delete Actions

from flo import ActionListOptions

# List all actions
actions = await client.action.list()

# List with prefix filter
actions = await client.action.list(ActionListOptions(prefix="process-"))

# Delete an action
await client.action.delete("process-image")

Worker Operations

Workers execute tasks for registered actions.

Register a Worker

# Register a worker that handles specific task types
await client.worker.register(
    "worker-1",
    ["process-image", "resize-image"]
)

Await and Process Tasks

from flo import WorkerAwaitOptions

# Wait for a task (blocking)
result = await client.worker.await_task(
    "worker-1",
    ["process-image", "resize-image"],
    WorkerAwaitOptions(block_ms=30000)  # Wait up to 30s
)

if result.task:
    task = result.task
    print(f"Got task: {task.task_id} for action: {task.action_name}")

    try:
        # Process the task
        output = process(task.input)

        # Complete successfully
        await client.worker.complete("worker-1", task.task_id, output)
    except Exception as e:
        # Fail the task (will be retried)
        await client.worker.fail("worker-1", task.task_id, str(e))

Extend Task Lease

from flo import WorkerTouchOptions

# For long-running tasks, extend the lease periodically
await client.worker.touch(
    "worker-1",
    task.task_id,
    WorkerTouchOptions(extend_ms=30000)
)

Worker Lifecycle Example

import asyncio
from flo import FloClient

async def run_worker():
    async with FloClient("localhost:9000", namespace="myapp") as client:
        # Create an action worker from the client
        worker = client.new_action_worker(concurrency=5)

        @worker.action("process-image")
        async def process_image(ctx):
            data = ctx.json()
            result = await do_processing(data)
            return ctx.to_bytes({"status": "done"})

        await worker.start()

asyncio.run(run_worker())

StreamWorker Example

import asyncio
from flo import FloClient, StreamContext

async def process_event(ctx: StreamContext) -> None:
    event = ctx.json()
    print(f"Got event: {event}")
    # Return normally → auto-ack
    # Raise an exception → auto-nack (redelivery)

async def run_stream_worker():
    async with FloClient("localhost:9000", namespace="myapp") as client:
        worker = client.new_stream_worker(
            stream="events",
            group="processors",
            handler=process_event,
            concurrency=5,
        )
        await worker.start()

asyncio.run(run_stream_worker())

Configuration

Client Options

client = FloClient(
    "localhost:9000",
    namespace="myapp",      # Default namespace for all operations
    timeout_ms=5000,        # Connection and operation timeout
    debug=True,             # Enable debug logging
)

Namespaces

Each operation can override the default namespace:

from flo import GetOptions, PutOptions, EnqueueOptions

# Use client's default namespace
await client.kv.put("key", b"value")

# Override namespace for this operation
await client.kv.put("key", b"value", PutOptions(namespace="other"))
await client.kv.get("key", GetOptions(namespace="other"))
await client.queue.enqueue("tasks", payload, EnqueueOptions(namespace="other"))

Error Handling

from flo import (
    FloError,
    NotFoundError,
    ConflictError,
    ConnectionFailedError,
)

try:
    await client.kv.put("key", b"new", PutOptions(cas_version=1))
except ConflictError:
    print("CAS version mismatch - value was modified")
except ConnectionFailedError:
    print("Failed to connect to server")
except FloError as e:
    print(f"Flo error: {e}")

Error Types

Error	Description
`NotFoundError`	Key or resource not found
`BadRequestError`	Invalid request parameters
`ConflictError`	CAS version mismatch
`UnauthorizedError`	Authentication failed
`OverloadedError`	Server is overloaded
`InternalServerError`	Server internal error
`ConnectionFailedError`	Failed to connect
`InvalidChecksumError`	Response CRC32 mismatch

Requirements

Python 3.10+
asyncio

License

MIT

Name		Name	Last commit message	Last commit date
.github/workflows		.github/workflows
examples		examples
src/flo		src/flo
tests		tests
.gitignore		.gitignore
LICENSE		LICENSE
README.md		README.md
pyproject.toml		pyproject.toml

Uh oh!

Latest commit

History

Folders and files

Repository files navigation

flo-python

Installation

Quick Start

Features

KV Operations

Get

Put

Delete

Scan

History

Queue Operations

Enqueue

Dequeue

Ack/Nack

Dead Letter Queue

Peek

Touch (Lease Renewal)

Stream Operations

Append

Read

Consumer Groups

Action Operations

Register an Action

Invoke an Action

Check Action Status

List and Delete Actions

Worker Operations

Register a Worker

Await and Process Tasks

Extend Task Lease

Worker Lifecycle Example

StreamWorker Example

Configuration

Client Options

Namespaces

Error Handling

Error Types

Requirements

License

About

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages