Health and Administration

The GQLDB Python driver provides methods for monitoring server health, managing caches, and gathering statistics.

Health Service Methods

Admin Service Methods

Task and Process Methods

Health Checks

health_check()

Check the current health status of the server:

Python
from gqldb import GqldbClient, GqldbConfig
from gqldb.types import HealthStatus

config = GqldbConfig(hosts=["localhost:9000"])

with GqldbClient(config) as client:
    client.login("admin", "password")

    # Check overall server health
    status = client.health_check()

    if status == HealthStatus.SERVING:
        print("Server is healthy and serving requests")
    elif status == HealthStatus.NOT_SERVING:
        print("Server is not serving requests")
    elif status == HealthStatus.SERVICE_UNKNOWN:
        print("Service status is unknown")
    else:
        print("Health status is unknown")

    # Check specific service health
    query_status = client.health_check("query")
    print(f"Query service: {query_status.name}")

HealthStatus Enum

Python
from gqldb.types import HealthStatus

HealthStatus.UNKNOWN         # Status unknown
HealthStatus.SERVING         # Healthy and serving
HealthStatus.NOT_SERVING     # Not serving requests
HealthStatus.SERVICE_UNKNOWN # Service status unknown

watch()

Monitor health status changes with a callback:

Python
from gqldb.types import HealthStatus

def on_health_change(status: HealthStatus):
    print(f"Health status changed: {status.name}")

    if status != HealthStatus.SERVING:
        print("WARNING: Server is not healthy!")
        # Trigger alerts, failover logic, etc.

# Watch health status
client.watch(callback=on_health_change)

watch_iter()

Iterate over health status changes:

Python
from gqldb.types import HealthStatus

# Using generator
for status in client.watch_iter():
    print(f"Health status: {status.name}")

    if status != HealthStatus.SERVING:
        print("Server unhealthy, breaking...")
        break

Cache Management

CacheType Enum

Python
from gqldb.types import CacheType

CacheType.ALL   # All caches
CacheType.AST   # Abstract Syntax Tree cache
CacheType.PLAN  # Query plan cache

get_cache_stats()

Get statistics about caches:

Python
from gqldb.types import CacheType

# Get all cache stats
all_stats = client.get_cache_stats(CacheType.ALL)
print(f"All cache stats: {all_stats}")

# Get AST cache stats
ast_stats = client.get_cache_stats(CacheType.AST)
print(f"AST cache: {ast_stats}")

# Get plan cache stats
plan_stats = client.get_cache_stats(CacheType.PLAN)
print(f"Plan cache: {plan_stats}")

CacheStats Class

Python
@dataclass
class CacheStats:
    ast_stats: ASTCacheStats
    plan_stats: PlanCacheStats

@dataclass
class ASTCacheStats:
    hits: int
    misses: int
    evictions: int
    entries: int
    hit_rate: float

@dataclass
class PlanCacheStats:
    hits: int
    misses: int
    size: int
    capacity: int
    hit_rate: float

clear_cache()

Clear caches to free memory or force recompilation:

Python
from gqldb.types import CacheType

# Clear all caches
client.clear_cache(CacheType.ALL)
print("All caches cleared")

# Clear only AST cache
client.clear_cache(CacheType.AST)
print("AST cache cleared")

# Clear only plan cache
client.clear_cache(CacheType.PLAN)
print("Plan cache cleared")

# Clear all (default)
client.clear_cache()
print("All caches cleared (default)")

Parser Warmup

warmup_parser()

Pre-allocate parser instances for better performance:

Python
# Pre-allocate 10 parser instances
client.warmup_parser(10)
print("Parsers warmed up")

This is useful before high-load periods to reduce latency from parser initialization.

Database Statistics

get_statistics()

Get statistics about the database or a specific graph:

Python
# Get overall database statistics
db_stats = client.get_statistics()
print(f"Database statistics: {db_stats}")

# Get statistics for a specific graph
graph_stats = client.get_statistics("myGraph")
print(f"Graph statistics: {graph_stats}")

Statistics Class

Python
@dataclass
class Statistics:
    node_count: int
    edge_count: int
    label_counts: Dict[str, int]
    edge_label_counts: Dict[str, int]

Permission Cache

invalidate_permission_cache()

Invalidate the RBAC (Role-Based Access Control) permission cache:

Python
# Invalidate all permission caches
client.invalidate_permission_cache()
print("All permission caches invalidated")

# Invalidate cache for a specific user
client.invalidate_permission_cache("johndoe")
print("Permission cache invalidated for johndoe")

Use this after changing user permissions to ensure changes take effect immediately.

System Metrics

get_system_metrics()

Get system-level metrics including CPU, memory, disk, and network:

Python
metrics = client.get_system_metrics()

if metrics.cpu:
    print(f"CPU - Process: {metrics.cpu.process_percent}%, System: {metrics.cpu.system_percent}%")

if metrics.memory:
    print(f"Memory - Used: {metrics.memory.system_used_percent}%")

if metrics.storage:
    print(f"Storage - DB size: {metrics.storage.db_size_bytes} bytes")

Compute Topology

wait_for_compute_topology()

Wait for the compute engine to be ready:

Python
ready, message = client.wait_for_compute_topology("myGraph", timeout=60)
print(f"Compute topology: {'ready' if ready else 'not ready'} - {message}")

Task Management

show_tasks()

List running and completed algorithm tasks:

Python
client.use_graph("myGraph")

tasks = client.show_tasks()
for t in tasks:
    print(f"{t.task_id}: {t.status} ({t.progress})")

stop_task() / delete_task()

Python
client.stop_task(task_id)
client.delete_task(task_id)

show_algos()

List available algorithms:

Python
algos = client.show_algos()
for a in algos:
    print(f"{a.name}: {a.description}")

Process Monitoring

top()

List currently running queries:

Python
procs = client.top()
for p in procs:
    print(f"Query {p.query_id}: {p.query_text} ({p.duration_ms}ms)")

kill()

Terminate a running query:

Python
client.kill(query_id)

stats()

Get graph statistics with node/edge counts by label:

Python
stats = client.stats()
print(f"Nodes: {stats.node_count}, Edges: {stats.edge_count}")
print(f"Label counts: {stats.label_counts}")

test()

Test connectivity (returns latency in nanoseconds):

Python
latency_ns = client.test()
print(f"Latency: {latency_ns / 1_000_000:.2f}ms")

Complete Example

Python
from gqldb import GqldbClient, GqldbConfig
from gqldb.types import CacheType, HealthStatus
from gqldb.errors import GqldbError
import time

def main():
    config = GqldbConfig(
        hosts=["localhost:9000"],
        timeout=30
    )

    with GqldbClient(config) as client:
        client.login("admin", "password")

        # Health check
        print("=== Health Check ===")
        health = client.health_check()
        print(f"Server status: {health.name}")

        # Warmup parsers
        print("\n=== Parser Warmup ===")
        client.warmup_parser(5)
        print("Warmed up 5 parser instances")

        # Execute some queries to populate cache
        print("\n=== Executing Queries ===")
        client.create_graph("healthDemo")
        client.use_graph("healthDemo")
        client.gql("MATCH (n) RETURN count(n)")
        client.gql("MATCH (n) RETURN count(n)")  # Should hit cache
        client.gql("MATCH (n)-[e]->(m) RETURN count(e)")

        # Check cache stats
        print("\n=== Cache Statistics ===")
        cache_stats = client.get_cache_stats(CacheType.ALL)
        print(f"Cache stats: {cache_stats}")

        # Get database statistics
        print("\n=== Database Statistics ===")
        db_stats = client.get_statistics()
        print(f"Database stats: {db_stats}")

        # Start health monitoring (with timeout)
        print("\n=== Health Monitoring ===")
        print("Monitoring health for 5 seconds...")

        start_time = time.time()
        for status in client.watch_iter():
            print(f"  Health update: {status.name}")
            if time.time() - start_time > 5:
                break

        print("Health monitoring stopped")

        # Clear caches
        print("\n=== Clear Caches ===")
        client.clear_cache(CacheType.ALL)
        print("All caches cleared")

        # Verify caches are cleared
        cleared_stats = client.get_cache_stats(CacheType.ALL)
        print(f"Cache stats after clear: {cleared_stats}")

        # Invalidate permission cache
        print("\n=== Permission Cache ===")
        client.invalidate_permission_cache()
        print("Permission cache invalidated")

        # Cleanup
        client.drop_graph("healthDemo")

if __name__ == "__main__":
    main()