graph-api-python 0.4.7

A fluent graph query library for Python with ElementStore and graph operations

Graph Elements Python Library

A Python library for graph-based data structures and queries, providing TypeScript-equivalent functionality for managing BaseElement, Node, Edge, MetaNode, and ElementStore classes with a powerful fluent GraphQuery API.

Features

• 🧩 Complete Graph Elements: BaseElement, Node, Edge, MetaNode classes with full TypeScript parity
• 🔍 Powerful Query API: Fluent GraphQuery interface with comprehensive filtering operators
• 🌐 REST API Client: Async HTTP client for remote graph servers with automatic serialization
• 📡 Real-time SSE Streaming: Server-Sent Events for live updates on elements, schemas, and queries
• 🔄 Synchronized Storage: Bi-directional sync between local cache and remote server via SSE
• 🧹 Clean API: Simple addNode(), createNode(), addEdge() methods with Props class
• 🔗 Shorter Query Syntax: classId(), where(), gte(), order_by(), first() aliases
• 🗄️ Flexible Storage: ElementStore with pluggable backends (memory, REST, synchronized)
• 📊 Rich Filtering: Support for EQ, GT, LT, GTE, LTE, BETWEEN, CONTAINS, STARTS_WITH, ENDS_WITH, REGEX, IN operators
• 🔄 Method Chaining: Intuitive fluent API for complex query composition
• 📈 Aggregation Functions: count, sum, mean/avg, median, min, max operations
• 🎯 Type Safe: Full typing support with TypeScript-equivalent interfaces
• ⚡ Async Support: Async operations throughout (nodes, edges, queries, streaming)
• ✅ Well Tested: Comprehensive test suite with 361 test cases

Installation

For End Users

# Install from PyPI (recommended)
pip install graph-api-python

# Or install from GitHub (latest development version)
pip install git+https://github.com/damylen/graph-api-python.git@main

# Install specific version from GitHub
pip install git+https://github.com/damylen/graph-api-python.git@v0.1.0

For Development

This project uses uv for fast, reliable Python package management.

# Install uv
curl -LsSf https://astral.sh/uv/install.sh | sh

# Clone and set up the project
git clone <repository-url>
cd graph-api
uv sync

Quick Start

from graph_api import GraphQuery, ElementStore, Props

# Create an element store
store = ElementStore()

# Add nodes using the clean API
alice = store.addNode('person', name="Alice", age=30, tags=["developer", "senior"])
bob = store.addNode('person', name="Bob", age=28, tags=["designer"])

# Or create with Props object
props = Props(name="Carol", age=32, department="Engineering")
carol = store.addNode('person', props)

# Create edges between nodes
friendship = store.addEdge('friendship', alice, bob, strength=0.9, type="close_friends")

# Create queries with intuitive syntax
query = GraphQuery(store)

# Basic filtering
people = query.classId('person').r()

# Advanced filtering with method chaining
senior_devs = (query
    .classId('person')
    .gte('age', 25)
    .contains('tags', 'senior')
    .r())

# Aggregations
total_people = query.classId('person').count()
avg_age = query.classId('person').avg('age')

# Sorting and limiting
youngest = (query
    .classId('person')
    .order_by('age', 'asc')
    .first(5)
    .r())

How It Works

Local Storage (In-Memory)

The ElementStore maintains an in-memory graph that you can query and modify locally. Perfect for standalone applications or testing.

REST API Integration

Connect to a remote TypeScript Graph API server (running on http://localhost:9800) to store and retrieve graph data:

• Automatic Serialization: Props objects and graph elements are automatically converted to/from JSON
• Async Operations: All REST calls are non-blocking
• Class-based Filtering: Query elements by their class ID efficiently

Real-Time Synchronization with SSE

Server-Sent Events stream live updates from the server:

• Synchronized Storage: Combine local ElementStore with SSE for instant updates on every server change
• Multiple Streams: Monitor elements, schemas, queries, and health status concurrently
• Conflict Resolution: Server-wins strategy for automatic conflict resolution

# Example: Synchronized storage with live updates
from graph_api import ElementStore, SynchronizedDataStorageConfig

sync_config = SynchronizedDataStorageConfig(
    base_url="http://localhost:9800",
    timeout=10.0
)
store = ElementStore(sync_config)

# Local operations are automatically synced to server via SSE
alice = store.addNode('person', name="Alice")

# Subscribe to live updates
async for event in store.stream_elements():
    print(f"Server updated: {event}")

## API Usage

The library provides a clean, intuitive API for graph operations:

### Node Operations

```python
from graph_api import ElementStore, Props

store = ElementStore()

# Add nodes with keyword arguments
person = store.addNode('person', name="John", age=30, city="Boston")

# Add nodes with Props object
props = Props(name="Alice", age=28, job="Engineer")
alice = store.addNode('person', props)

# Create nodes without adding to store
node = store.createNode('person', name="Bob", age=25)
# Later add to store
store.addNode(node)

# Async versions available
async def add_user():
    user = await store.addNodeAsync('user', name="Async User", active=True)
    return user

Edge Operations

# Create relationships between nodes
alice = store.addNode('person', name="Alice")
bob = store.addNode('person', name="Bob")

# Add edge with properties
friendship = store.addEdge('friendship', alice, bob,
                          strength=0.9, type="close_friends")

# Create edge without adding to store
edge = store.createEdge('relationship', alice, bob,
                       Props(type="colleague", department="Engineering"))

# Async edge creation
async def create_relationship():
    rel = await store.addEdgeAsync('follows', alice, bob, since="2023")
    return rel

Props Class

The Props class provides a simple way to define element properties:

from graph_api import Props

# Create props with keyword arguments
props = Props(name="Alice", age=30, job="Engineer")

# Props behaves like a dictionary
props['department'] = "Engineering"
assert props['name'] == "Alice"
assert 'age' in props

Query Methods

Filtering Methods

• classId(class_id): Filter by element class ID
• where(key, operator, value): Filter by property with operator
• where(key, value): Filter by property equality (shorthand)
• prop(key, value): Alias for property equality
• gt(key, value), lt(key, value), gte(key, value), lte(key, value): Comparison operators
• contains(key, value): Check if property contains value
• startsWith(key, value), endsWith(key, value): String matching

Sorting and Limiting

• order_by(property, direction): Sort by property ('asc' or 'desc')
• first(n): Limit to first N results
• last(n): Limit to last N results

Aggregation Methods

• count(): Count results
• sum(property): Sum numeric property
• avg(property): Average of numeric property
• median(property): Median of numeric property
• min(property): Minimum of numeric property
• max(property): Maximum of numeric property

REST Client API

Connect to a TypeScript Graph API server for remote graph operations. The client handles serialization automatically.

Configuration

from graph_api import RestGraphApiClient, RestClientConfig

# Configure the client
config = RestClientConfig(
    base_url="http://localhost:9800",
    timeout=30.0,
    verify_ssl=True
)

# Use the client
async with RestGraphApiClient(config) as client:
    # Get all elements
    all_elements = await client.get_all_elements()

    # Get single element by ID
    element = await client.get_element_by_id("element-id")

    # Get elements by class ID (NEW FEATURE)
    result = await client.get_elements_by_class_id("Person")
    print(f"Found {result['count']} Person elements")
    for person in result['data']:
        print(f"  - {person['properties'].get('name', 'Unknown')}")

    # Legacy method (returns list only)
    persons = await client.get_elements_by_class("Person")

    # Create/update elements
    new_elements = [
        {
            "id": "person-1",
            "classId": "Person",
            "type": "node",
            "properties": {"name": "Alice", "age": 30},
            "source": "python"
        }
    ]
    created = await client.create_elements(new_elements)

    # Query elements
    query_result = await client.query_elements({
        "classId": "Person",
        "filters": [{"property": "age", "operator": "gt", "value": 25}]
    })

REST Data Storage Integration

Use REST backend with ElementStore for seamless remote data access:

from graph_api import ElementStore, ElementStoreConfig
from graph_api import RestDataStorage, RestDataStorageConfig

# Configure REST storage
rest_config = RestDataStorageConfig(
    base_url="http://localhost:9800",
    timeout=30.0
)

# Create REST storage instance
rest_storage = RestDataStorage(rest_config.to_rest_config())

# Configure ElementStore with REST backend
store_config = ElementStoreConfig(
    data_config=rest_config,
    operation=rest_storage
)

# Create ElementStore
store = ElementStore(store_config)

# Load data from remote API
await store.load_elements()

# Use normal ElementStore methods - data is automatically synced with API
person = store.addNode('Person', name="Bob", age=35)

# Get elements by class ID from remote storage
persons = await store.get_elements_by_class_id("Person")
print(f"Found {len(persons)} persons in remote storage")

# Direct REST storage access
filtered_elements = await rest_storage.get_elements_by_class_id("Person")

Error Handling

The REST client provides specific exception types for different error conditions:

from graph_api import (
    GraphApiError, ValidationError, ElementNotFoundError,
    QueryError, InternalServerError, HttpError
)

try:
    result = await client.get_elements_by_class_id("NonExistentClass")
except ValidationError as e:
    print(f"Validation error: {e.message}")
except ElementNotFoundError as e:
    print(f"No elements found: {e.message}")
except InternalServerError as e:
    print(f"Server error: {e.message}")
except HttpError as e:
    print(f"HTTP error: {e.message}")
except GraphApiError as e:
    print(f"General API error: {e.message}")

Server-Sent Events (SSE) Integration

Receive real-time updates from the server using streaming. Use with synchronized storage for automatic local cache updates.

Streaming Methods

from graph_api import RestGraphApiClient

async with RestGraphApiClient() as client:
    # Stream all elements in real-time
    async for event in client.stream_elements():
        print(f"Element event: {event}")
        # Handle element creation, updates, deletions

    # Stream specific element changes
    async for event in client.stream_element_by_id("person-123"):
        print(f"Person 123 updated: {event}")

    # Stream elements by class ID
    async for event in client.stream_elements_by_class_id("Person"):
        print(f"Person element changed: {event}")

    # Stream schema changes
    async for event in client.stream_schema():
        print(f"Schema updated: {event}")

    # Stream continuous query results
    query = {"classId": "Person", "filters": [{"property": "active", "value": True}]}
    async for event in client.stream_query(query):
        print(f"Active person changed: {event}")

    # Stream health status
    async for event in client.stream_health():
        print(f"Health status: {event}")

    # Stream all graph events
    async for event in client.stream_events():
        print(f"Graph event: {event}")

SSE Event Format

SSE events are JSON objects containing event data:

{
  "type": "element_created",
  "elementId": "person-123",
  "data": {
    "id": "person-123",
    "classId": "Person",
    "type": "node",
    "properties": {"name": "Alice", "age": 30}
  },
  "timestamp": "2024-01-15T10:30:00Z"
}

Concurrent Streaming

Monitor multiple streams simultaneously:

import asyncio

async def monitor_elements():
    async with RestGraphApiClient() as client:
        # Create concurrent monitoring tasks
        tasks = [
            asyncio.create_task(monitor_stream("Elements", client.stream_elements())),
            asyncio.create_task(monitor_stream("Health", client.stream_health())),
            asyncio.create_task(monitor_stream("Schema", client.stream_schema())),
        ]

        # Run for a limited time or until cancelled
        await asyncio.wait_for(asyncio.gather(*tasks), timeout=300.0)

async def monitor_stream(name: str, stream):
    """Monitor a specific stream."""
    count = 0
    async for event in stream:
        count += 1
        print(f"{name} Event {count}: {event}")
        if count >= 10:  # Limit events per stream
            break

SSE Error Handling

SSE streams include built-in error handling:

try:
    async for event in client.stream_elements():
        process_event(event)
except HttpError as e:
    print(f"SSE connection failed: {e}")
except Exception as e:
    print(f"SSE processing error: {e}")

SSE Connectivity Testing

Check if SSE endpoints are available:

from graph_api.examples.sse_examples import sse_connectivity_test

# Run connectivity test
await sse_connectivity_test()

Example Usage

# Chain methods for complex queries
young_engineers = (query
    .classId('person')
    .where('department', 'Engineering')
    .lt('age', 35)
    .order_by('age', 'desc')
    .first(10)
    .r())

# Use comparison operators
seniors = query.classId('person').gte('age', 30).r()

# Aggregations
avg_age = query.classId('person').avg('age')
total_people = query.classId('person').count()

Development

This project uses uv for dependency management and development workflows.

Prerequisites

• Python 3.12+
• uv

Setup

# Install uv if you haven't already
curl -LsSf https://astral.sh/uv/install.sh | sh

# Clone the repository
git clone <repository-url>
cd graph-api

# Create virtual environment and install dependencies
uv sync

# Install in development mode
uv pip install -e .

Running Tests

# Run all tests
uv run pytest tests/ -v

# Run tests with coverage
uv run pytest tests/ -v --cov=graph_api --cov-report=html

# Run specific test file
uv run pytest tests/test_graph_query.py -v

Code Quality

# Run linting
uv run ruff check .

# Format code
uv run black .

# Sort imports
uv run isort .

# Type checking
uv run mypy graph_api/

# All quality checks
uv run ruff check . && uv run black --check . && uv run isort --check-only . && uv run mypy graph_api/

Adding Dependencies

# Add runtime dependency
uv add requests

# Add development dependency
uv add --dev pytest-mock

# Update dependencies
uv lock --upgrade

Using Makefile (Optional)

For convenience, common development tasks are available via Makefile:

# Set up development environment
make dev

# Run all quality checks and tests
make qa

# Run tests with coverage
make test

# Format code and fix linting
make lint-fix

# Build package
make build

# See all available commands
make help

Test Coverage

• ✅ 297 tests passing
• ✅ All filtering operators tested
• ✅ Method chaining validated
• ✅ Edge cases covered
• ✅ Aggregation functions verified
• ✅ Clean API methods tested
• ✅ Async operations validated

License

MIT License

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.