Python client SDK for the Fila message broker.
pip install fila-pythonfromfilaimportClientclient=Client("localhost:5555")
# Enqueue a message.msg_id=client.enqueue("my-queue", {"tenant": "acme"}, b"hello world")
print(f"Enqueued: {msg_id}")
# Consume messages.formsginclient.consume("my-queue"):
print(f"Received: {msg.id} (attempt {msg.attempt_count})")
# Acknowledge successful processing.client.ack("my-queue", msg.id)
client.close()importasynciofromfilaimportAsyncClientasyncdefmain():
asyncwithAsyncClient("localhost:5555") asclient:
msg_id=awaitclient.enqueue("my-queue", {"tenant": "acme"}, b"hello")
asyncformsginawaitclient.consume("my-queue"):
print(f"Received: {msg.id}")
awaitclient.ack("my-queue", msg.id)
asyncio.run(main())For servers using certificates from a public CA (e.g., Let's Encrypt), enable TLS with the system trust store:
fromfilaimportClient# TLS using OS system trust store.client=Client("localhost:5555", tls=True)For servers using a private CA, provide the CA certificate explicitly:
fromfilaimportClient# Read certificates.withopen("ca.pem", "rb") asf:
ca_cert=f.read()
withopen("client.pem", "rb") asf:
client_cert=f.read()
withopen("client-key.pem", "rb") asf:
client_key=f.read()
# TLS with custom CA (server verification).client=Client("localhost:5555", ca_cert=ca_cert)
# Mutual TLS (client + server verification).client=Client(
"localhost:5555",
ca_cert=ca_cert,
client_cert=client_cert,
client_key=client_key,
)Note: ca_cert implies tls=True -- you don't need to pass both.
When the server has API key auth enabled, pass the key to the client:
fromfilaimportClientclient=Client("localhost:5555", api_key="fila_your_api_key_here")
# Combined with TLS:client=Client(
"localhost:5555",
ca_cert=ca_cert,
api_key="fila_your_api_key_here",
)The API key is sent as authorization: Bearer <key> metadata on every RPC.
Client(addr, *, tls=False, ca_cert=None, client_cert=None, client_key=None, api_key=None) / AsyncClient(...)
Connect to a Fila broker. Both support context manager protocol.
Enqueue a message. Returns the broker-assigned message ID.
Open a streaming consumer. Returns an iterator (sync) or async iterator (async) that yields messages as they become available.
Acknowledge a successfully processed message. The message is permanently removed.
Negatively acknowledge a failed message. The message is requeued or routed to the dead-letter queue based on the queue's configuration.
Per-operation exception classes:
fromfilaimportQueueNotFoundError, MessageNotFoundErrortry:
client.enqueue("missing-queue", None, b"test")
exceptQueueNotFoundError:
print("Queue does not exist")
try:
client.ack("my-queue", "missing-id")
exceptMessageNotFoundError:
print("Message does not exist")AGPLv3