falkordb-py

FalkorDB Python client

see docs

Installation

pip install FalkorDB

For faster parsing of large results (for example vector properties), install the optional hiredis parser. redis uses it automatically:

pip install "FalkorDB[hiredis]"

Usage

Run FalkorDB instance

Docker:

docker run --rm -p 6379:6379 falkordb/falkordb

Or use FalkorDB Cloud

Synchronous Example

from falkordb import FalkorDB

# Connect to FalkorDB
db = FalkorDB(host="localhost", port=6379)

# Select the social graph
g = db.select_graph("social")

# Create 100 nodes and return a handful
nodes = g.query(
    "UNWIND range(0, 100) AS i CREATE (n {v:1}) RETURN n LIMIT 10"
).result_set
for n in nodes:
    print(n)

# Read-only query the graph for the first 10 nodes
nodes = g.ro_query("MATCH (n) RETURN n LIMIT 10").result_set

# Copy the Graph
copy_graph = g.copy("social_copy")

# Delete the Graph
g.delete()

Asynchronous Example

import asyncio
from falkordb.asyncio import FalkorDB
from redis.asyncio import BlockingConnectionPool


async def main():

    # Connect to FalkorDB
    pool = BlockingConnectionPool(
        max_connections=16, timeout=None, decode_responses=True
    )
    db = FalkorDB(connection_pool=pool)

    # Select the social graph
    g = db.select_graph("social")

    # Execute query asynchronously
    result = await g.query(
        "UNWIND range(0, 100) AS i CREATE (n {v:1}) RETURN n LIMIT 10"
    )

    # Process results
    for n in result.result_set:
        print(n)

    # Run multiple queries concurrently
    tasks = [
        g.query("MATCH (n) WHERE n.v = 1 RETURN count(n) AS count"),
        g.query('CREATE (p:Person {name: "Alice"}) RETURN p'),
        g.query('CREATE (p:Person {name: "Bob"}) RETURN p'),
    ]

    results = await asyncio.gather(*tasks)

    # Process concurrent results
    print(f"Node count: {results[0].result_set[0][0]}")
    print(f"Created Alice: {results[1].result_set[0][0]}")
    print(f"Created Bob: {results[2].result_set[0][0]}")

    # Close the connection when done
    await pool.aclose()


# Run the async example
if __name__ == "__main__":
    asyncio.run(main())

Sentinel Failover

Point the client at a Sentinel and it connects to the monitored master. The Sentinel must monitor exactly one master; the client raises an error otherwise. After a failover, it reconnects to the new master:

from falkordb import FalkorDB

db = FalkorDB(host="localhost", port=26379)

A dropped connection or a timeout does not tell whether a write already ran on the server, so a write that fails this way is not sent again: the error is raised to the caller. Reads, and writes that provably never ran (the connection could not be opened, or a demoted master rejected the write as read-only), are retried according to the client's retry settings. FalkorDB does not retry by default, so pass a retry policy to ride out a failover.

To resend writes as well, pass retry_writes=True. A resent write may then run twice, so only enable it when your write queries are safe to repeat, for example MERGE instead of CREATE:

from redis.backoff import ExponentialBackoff
from redis.exceptions import ConnectionError, ReadOnlyError, TimeoutError
from redis.retry import Retry

db = FalkorDB(
    host="localhost",
    port=26379,
    retry=Retry(ExponentialBackoff(cap=1, base=0.1), 15),
    retry_on_error=[ConnectionError, TimeoutError, ReadOnlyError],
    retry_writes=True,
)