FalkorDriver

FalkorDB graph database driver implementation

The FalkorDriver class provides connectivity to FalkorDB, a Redis-based graph database with RedisSearch integration.

Overview

FalkorDB is a multi-tenant graph database built on Redis. The FalkorDriver provides:

• Redis-based graph storage
• RedisSearch fulltext indexing
• Multi-tenant database support
• Connection to FalkorDB Cloud or on-premises
• Automatic datetime conversion

Constructor

from graphiti_core.driver.falkordb_driver import FalkorDriver

driver = FalkorDriver(
    host="localhost",
    port=6379,
    username=None,
    password=None,
    database="default_db"
)

Parameters

host · str · default: "localhost"

The hostname where FalkorDB is running.

For on-premises deployments, this is typically "localhost". For FalkorDB Cloud, use the provided cloud endpoint.

Example: "localhost" or "cloud.falkordb.com"

port · int · default: "6379"

The port on which FalkorDB is listening.

The default Redis port is 6379. FalkorDB Cloud may use different ports.

Example: 6379

username · str | None · default: "None"

The username for authentication.

Optional for on-premises deployments. Required for FalkorDB Cloud.

Example: "admin"

password · str | None · default: "None"

The password for authentication.

Optional for on-premises deployments. Required for FalkorDB Cloud.

Example: "your-password"

falkor_db · FalkorDB | None · default: "None"

An existing FalkorDB client instance to reuse.

If provided, the driver will use this client instead of creating a new one. Useful for sharing connections or custom client configurations.

Example: FalkorDB(host="localhost", port=6379)

database · str · default: "default_db"

The name of the database (graph) to connect to.

FalkorDB supports multiple graphs. Each graph is isolated.

Example: "graphiti" or "user_graph"

Methods

execute_query

Execute a Cypher query and return results.

result = await driver.execute_query(
    "MATCH (n:Entity) RETURN n LIMIT 10"
)

Parameters:

• cypher_query_ (str) - The Cypher query to execute
• **kwargs - Query parameters (automatically converted for FalkorDB)

Returns: tuple[list[dict], list[str], None] - Records, headers, and summary

FalkorDB does not support native datetime objects. The driver automatically converts datetime values to ISO format strings.

session

Create a new database session.

session = driver.session(database="graphiti")
try:
    result = await session.run("MATCH (n) RETURN count(n)")
finally:
    await session.close()

Parameters:

• database (str | None) - Optional database name override

Returns: FalkorDriverSession - Session object

close

Close the driver and release connections.

await driver.close()

Returns: None

build_indices_and_constraints

Create all required indices for Graphiti.

await driver.build_indices_and_constraints(delete_existing=False)

Parameters:

• delete_existing (bool) - If True, delete existing indices before creating new ones

Returns: None

This method is automatically called during driver initialization.

health_check

Verify connectivity to FalkorDB.

try:
    await driver.health_check()
    print("Connection healthy")
except Exception as e:
    print(f"Connection failed: {e}")

Returns: None - Raises exception if connection fails

clone

Create a copy of the driver targeting a different database.

default_driver = FalkorDriver(host="localhost", database="default_db")
graphs_driver = default_driver.clone("user_graphs")

Parameters:

• database (str) - The name of the database to target

Returns: GraphDriver - New driver instance

build_fulltext_query

Build a RedisSearch-compatible fulltext query string.

query = driver.build_fulltext_query(
    query="California Attorney General",
    group_ids=["main", "politics"],
    max_query_length=128
)
# Returns: '(@group_id:"main"|"politics") (California | Attorney | General)'

Parameters:

• query (str) - The search query text
• group_ids (list[str] | None) - Optional list of group IDs to filter by
• max_query_length (int) - Maximum query length (default: 128)

Returns: str - RedisSearch formatted query

FalkorDB uses RedisSearch syntax with the @ prefix for field queries. The driver automatically formats queries and escapes reserved characters.

sanitize

Sanitize query text by replacing FalkorDB special characters with whitespace.

sanitized = driver.sanitize("[e-mail], test-query!")
# Returns: "user example com  test query "

Parameters:

• query (str) - The text to sanitize

Returns: str - Sanitized text with special characters replaced

Properties

provider

driver.provider  # GraphProvider.FALKORDB

Returns the GraphProvider.FALKORDB enum value.

fulltext_syntax

driver.fulltext_syntax  # '@'

The prefix used for fulltext field queries in RedisSearch syntax.

default_group_id

driver.default_group_id  # '\\_'

The default group ID used for FalkorDB queries.

Operations Properties

The driver exposes specialized operation interfaces:

driver.entity_node_ops       # FalkorEntityNodeOperations
driver.episode_node_ops      # FalkorEpisodeNodeOperations
driver.community_node_ops    # FalkorCommunityNodeOperations
driver.saga_node_ops         # FalkorSagaNodeOperations
driver.entity_edge_ops       # FalkorEntityEdgeOperations
driver.episodic_edge_ops     # FalkorEpisodicEdgeOperations
driver.community_edge_ops    # FalkorCommunityEdgeOperations
driver.has_episode_edge_ops  # FalkorHasEpisodeEdgeOperations
driver.next_episode_edge_ops # FalkorNextEpisodeEdgeOperations
driver.search_ops            # FalkorSearchOperations
driver.graph_ops             # FalkorGraphMaintenanceOperations

Usage Examples

Basic Connection (On-Premises)

import asyncio
from graphiti_core.driver.falkordb_driver import FalkorDriver

async def main():
    # Connect to local FalkorDB
    driver = FalkorDriver(
        host="localhost",
        port=6379
    )
    
    try:
        # Verify connection
        await driver.health_check()
        print("Connected to FalkorDB")
        
        # Execute a query
        result, headers, _ = await driver.execute_query(
            "MATCH (n:Entity) RETURN count(n) as count"
        )
        print(f"Entity count: {result[0]['count']}")
    finally:
        await driver.close()

asyncio.run(main())

FalkorDB Cloud Connection

import os
from dotenv import load_dotenv
from graphiti_core.driver.falkordb_driver import FalkorDriver

load_dotenv()

driver = FalkorDriver(
    host=os.environ.get("FALKORDB_HOST", "localhost"),
    port=int(os.environ.get("FALKORDB_PORT", "6379")),
    username=os.environ.get("FALKORDB_USERNAME"),
    password=os.environ.get("FALKORDB_PASSWORD"),
    database="graphiti"
)

With Graphiti

from graphiti_core import Graphiti
from graphiti_core.driver.falkordb_driver import FalkorDriver
from graphiti_core.nodes import EpisodeType
from datetime import datetime, timezone
import json

# Create driver
driver = FalkorDriver(
    host="localhost",
    port=6379,
    database="graphiti"
)

# Initialize Graphiti
graphiti = Graphiti(graph_driver=driver)

# Build indices
await graphiti.build_indices_and_constraints()

# Add an episode
await graphiti.add_episode(
    name="Episode 1",
    episode_body="Kamala Harris is the Attorney General of California.",
    source=EpisodeType.text,
    source_description="biographical information",
    reference_time=datetime.now(timezone.utc)
)

# Search
results = await graphiti.search(
    "Who is the California Attorney General?"
)
print(results)

Multi-Tenant Usage

from graphiti_core.driver.falkordb_driver import FalkorDriver

# Create a base driver
base_driver = FalkorDriver(
    host="localhost",
    port=6379
)

# Create drivers for different tenants (graphs)
user1_driver = base_driver.clone("user_1_graph")
user2_driver = base_driver.clone("user_2_graph")
shared_driver = base_driver.clone("shared_graph")

# Each driver operates on its own isolated graph
await user1_driver.execute_query(
    "CREATE (n:Entity {name: 'User 1 Entity'})"
)
await user2_driver.execute_query(
    "CREATE (n:Entity {name: 'User 2 Entity'})"
)

# Queries are isolated by graph
result1, _, _ = await user1_driver.execute_query(
    "MATCH (n:Entity) RETURN n.name"
)  # Returns only User 1 entities

result2, _, _ = await user2_driver.execute_query(
    "MATCH (n:Entity) RETURN n.name"
)  # Returns only User 2 entities

Fulltext Search with RedisSearch

from graphiti_core.driver.falkordb_driver import FalkorDriver

driver = FalkorDriver(host="localhost")

# Build a RedisSearch query
query = driver.build_fulltext_query(
    query="California politics government",
    group_ids=["politics", "history"]
)

print(query)
# Output: (@group_id:"politics"|"history") (California | politics | government)

# Use in a search
result, _, _ = await driver.execute_query(
    f"""
    CALL db.idx.fulltext.queryNodes('entities', '{query}')
    YIELD node
    RETURN node
    LIMIT 10
    """
)

Reusing an Existing FalkorDB Client

from falkordb.asyncio import FalkorDB
from graphiti_core.driver.falkordb_driver import FalkorDriver

# Create a custom FalkorDB client
falkor_client = FalkorDB(
    host="localhost",
    port=6379,
    username="admin",
    password="password"
)

# Reuse it in multiple drivers
driver1 = FalkorDriver(falkor_db=falkor_client, database="graph1")
driver2 = FalkorDriver(falkor_db=falkor_client, database="graph2")

# Both drivers share the same connection pool

Environment Variables

Common environment variables for FalkorDB configuration:

FALKORDB_HOST=localhost
FALKORDB_PORT=6379
FALKORDB_USERNAME=admin
FALKORDB_PASSWORD=your-password

Load in your application:

import os
from dotenv import load_dotenv
from graphiti_core.driver.falkordb_driver import FalkorDriver

load_dotenv()

driver = FalkorDriver(
    host=os.environ.get("FALKORDB_HOST", "localhost"),
    port=int(os.environ.get("FALKORDB_PORT", "6379")),
    username=os.environ.get("FALKORDB_USERNAME"),
    password=os.environ.get("FALKORDB_PASSWORD")
)

Installation

FalkorDB support requires the optional falkordb extra:

pip install graphiti-core[falkordb]

Or install the FalkorDB client directly:

pip install falkordb

Datetime Handling

FalkorDB does not support native datetime objects. The driver automatically converts:

from datetime import datetime, timezone

# Datetime objects are automatically converted to ISO strings
await driver.execute_query(
    "CREATE (e:Episode {created_at: $created_at})",
    created_at=datetime.now(timezone.utc)
)
# Stored as: "2024-03-15T10:30:00+00:00"

The conversion happens automatically in:

• Query parameters
• List values
• Nested dictionary values

Stopwords

FalkorDB’s RedisSearch automatically filters common stopwords. The driver includes a comprehensive stopword list that mirrors RedisSearch behavior:

from graphiti_core.driver.falkordb import STOPWORDS

print(len(STOPWORDS))  # 604 stopwords

Stopwords are automatically removed when building fulltext queries.

Related

Drivers Overview

Learn about the driver architecture

Graphiti

Main Graphiti class documentation

Neo4j Driver

Alternative production-ready driver

Kuzu Driver

Embedded analytics driver