tessl/pypi-python-arango

Python Driver for ArangoDB, a scalable multi-model database that natively supports documents, graphs, and search.

—

Pending

Overview

Eval results

Files

Graph Operations

Name: tessl/pypi-python-arango
Author: tessl

Comprehensive graph database functionality for managing vertices, edges, and graph traversals in ArangoDB. Supports named graphs with flexible edge definitions, graph collections, and specialized graph operations.

Capabilities

Graph Management

Create, delete, and manage named graphs with configurable edge definitions and vertex collections.

class StandardDatabase:
    def graph(self, name: str) -> Graph:
        """
        Get graph interface.
        
        Parameters:
        - name: str, graph name
        
        Returns:
        Graph: Graph interface object
        """
        
    def graphs(self) -> Result[List[Json]]:
        """
        List all graphs.
        
        Returns:
        Result[List[Json]]: List of graph information dicts
        """
        
    def create_graph(self, name: str, edge_definitions=None,
                     orphan_collections=None, **kwargs) -> Result[Graph]:
        """
        Create a new graph.
        
        Parameters:
        - name: str, graph name
        - edge_definitions: list, edge collection definitions
        - orphan_collections: list, standalone vertex collections
        - **kwargs: additional graph options
        
        Returns:
        Result[Graph]: Created graph object
        """
        
    def delete_graph(self, name: str, ignore_missing: bool = False,
                     drop_collections=None) -> Result[bool]:
        """
        Delete a graph.
        
        Parameters:
        - name: str, graph name
        - ignore_missing: bool, ignore if graph doesn't exist
        - drop_collections: bool, delete associated collections
        
        Returns:
        Result[bool]: True on success
        """
        
    def has_graph(self, name: str) -> Result[bool]:
        """
        Check if graph exists.
        
        Parameters:
        - name: str, graph name
        
        Returns:
        Result[bool]: True if graph exists
        """

Graph Properties

Access graph configuration, edge definitions, and collection information.

class Graph:
    @property
    def name(self) -> str:
        """Graph name."""
        
    def properties(self) -> Result[Json]:
        """
        Get graph properties.
        
        Returns:
        Result[Json]: Graph configuration dict
        """

Vertex Collection Management

Manage vertex collections within graphs with creation, deletion, and listing operations.

def vertex_collections(self) -> Result[List[str]]:
    """
    List vertex collection names.
    
    Returns:
    Result[List[str]]: List of vertex collection names
    """
    
def has_vertex_collection(self, name: str) -> Result[bool]:
    """
    Check if vertex collection exists in graph.
    
    Parameters:
    - name: str, collection name
    
    Returns:
    Result[bool]: True if collection exists in graph
    """
    
def vertex_collection(self, name: str) -> VertexCollection:
    """
    Get vertex collection interface.
    
    Parameters:
    - name: str, collection name
    
    Returns:
    VertexCollection: Vertex collection object
    """
    
def create_vertex_collection(self, name: str, options=None) -> Result[VertexCollection]:
    """
    Create vertex collection in graph.
    
    Parameters:
    - name: str, collection name
    - options: dict, collection creation options
    
    Returns:
    Result[VertexCollection]: Created vertex collection
    """
    
def delete_vertex_collection(self, name: str, purge: bool = False) -> Result[bool]:
    """
    Remove vertex collection from graph.
    
    Parameters:
    - name: str, collection name
    - purge: bool, delete collection completely
    
    Returns:
    Result[bool]: True on success
    """

Edge Definition Management

Define and manage edge collections with their vertex collection relationships.

def edge_definitions(self) -> Result[List[Json]]:
    """
    List edge definitions.
    
    Returns:
    Result[List[Json]]: List of edge definition dicts
    """
    
def has_edge_definition(self, name: str) -> Result[bool]:
    """
    Check if edge definition exists.
    
    Parameters:
    - name: str, edge collection name
    
    Returns:
    Result[bool]: True if edge definition exists
    """
    
def has_edge_collection(self, name: str) -> Result[bool]:
    """
    Check if edge collection exists in graph.
    
    Parameters:
    - name: str, collection name
    
    Returns:
    Result[bool]: True if collection exists
    """
    
def edge_collection(self, name: str) -> EdgeCollection:
    """
    Get edge collection interface.
    
    Parameters:
    - name: str, collection name
    
    Returns:
    EdgeCollection: Edge collection object
    """
    
def create_edge_definition(self, edge_collection: str,
                           from_vertex_collections: Sequence[str],
                           to_vertex_collections: Sequence[str],
                           options=None) -> Result[EdgeCollection]:
    """
    Create edge definition.
    
    Parameters:
    - edge_collection: str, edge collection name
    - from_vertex_collections: list, source vertex collection names
    - to_vertex_collections: list, target vertex collection names
    - options: dict, edge collection options
    
    Returns:
    Result[EdgeCollection]: Created edge collection
    """
    
def replace_edge_definition(self, edge_collection: str,
                            from_vertex_collections: Sequence[str],
                            to_vertex_collections: Sequence[str],
                            sync=None, purge: bool = False) -> Result[EdgeCollection]:
    """
    Replace edge definition.
    
    Parameters:
    - edge_collection: str, edge collection name
    - from_vertex_collections: list, source vertex collection names
    - to_vertex_collections: list, target vertex collection names
    - sync: bool, wait for sync to disk
    - purge: bool, purge orphaned edges
    
    Returns:
    Result[EdgeCollection]: Updated edge collection
    """
    
def delete_edge_definition(self, name: str, purge: bool = False,
                           sync=None) -> Result[bool]:
    """
    Delete edge definition.
    
    Parameters:
    - name: str, edge collection name
    - purge: bool, delete collection completely
    - sync: bool, wait for sync to disk
    
    Returns:
    Result[bool]: True on success
    """
    
def edge_collections(self) -> Result[List[str]]:
    """
    List edge collection names.
    
    Returns:
    Result[List[str]]: List of edge collection names
    """

Vertex Operations

Direct vertex manipulation within graph context with validation of graph constraints.

def has_vertex(self, vertex, rev=None, check_rev: bool = True) -> Result[bool]:
    """
    Check if vertex exists.
    
    Parameters:
    - vertex: str or dict, vertex identifier
    - rev: str, expected revision
    - check_rev: bool, check revision matches
    
    Returns:
    Result[bool]: True if vertex exists
    """
    
def vertex(self, vertex, rev=None, check_rev: bool = True) -> Result:
    """
    Get vertex document.
    
    Parameters:
    - vertex: str or dict, vertex identifier
    - rev: str, expected revision
    - check_rev: bool, check revision matches
    
    Returns:
    Result[Json]: Vertex document or None
    """
    
def insert_vertex(self, collection: str, vertex: Json, sync=None) -> Result[Json]:
    """
    Insert vertex into graph.
    
    Parameters:
    - collection: str, vertex collection name
    - vertex: dict, vertex document data
    - sync: bool, wait for sync to disk
    
    Returns:
    Result[Json]: Vertex document metadata
    """
    
def update_vertex(self, vertex: Json, check_rev: bool = True,
                  keep_none: bool = True, sync=None) -> Result[Json]:
    """
    Update vertex in graph.
    
    Parameters:
    - vertex: dict, vertex document with _key
    - check_rev: bool, check current revision
    - keep_none: bool, keep null values
    - sync: bool, wait for sync to disk
    
    Returns:
    Result[Json]: Updated vertex metadata
    """
    
def replace_vertex(self, vertex: Json, check_rev: bool = True,
                   sync=None) -> Result[Json]:
    """
    Replace vertex in graph.
    
    Parameters:
    - vertex: dict, complete replacement vertex
    - check_rev: bool, check current revision
    - sync: bool, wait for sync to disk
    
    Returns:
    Result[Json]: Replaced vertex metadata
    """
    
def delete_vertex(self, vertex, rev=None, check_rev: bool = True,
                  ignore_missing: bool = False, sync=None) -> Result:
    """
    Delete vertex from graph.
    
    Parameters:
    - vertex: str or dict, vertex identifier
    - rev: str, expected revision
    - check_rev: bool, check revision matches
    - ignore_missing: bool, ignore if vertex doesn't exist
    - sync: bool, wait for sync to disk
    
    Returns:
    Result[Json|bool]: Deletion metadata or boolean
    """

Edge Operations

Direct edge manipulation with automatic validation of vertex collection constraints.

def has_edge(self, edge, rev=None, check_rev: bool = True) -> Result[bool]:
    """
    Check if edge exists.
    
    Parameters:
    - edge: str or dict, edge identifier
    - rev: str, expected revision
    - check_rev: bool, check revision matches
    
    Returns:
    Result[bool]: True if edge exists
    """
    
def edge(self, edge, rev=None, check_rev: bool = True) -> Result:
    """
    Get edge document.
    
    Parameters:
    - edge: str or dict, edge identifier
    - rev: str, expected revision
    - check_rev: bool, check revision matches
    
    Returns:
    Result[Json]: Edge document or None
    """
    
def insert_edge(self, collection: str, edge: Json, sync=None) -> Result[Json]:
    """
    Insert edge into graph.
    
    Parameters:
    - collection: str, edge collection name
    - edge: dict, edge document with _from and _to
    - sync: bool, wait for sync to disk
    
    Returns:
    Result[Json]: Edge document metadata
    """
    
def update_edge(self, edge: Json, check_rev: bool = True,
                keep_none: bool = True, sync=None) -> Result[Json]:
    """
    Update edge in graph.
    
    Parameters:
    - edge: dict, edge document with _key
    - check_rev: bool, check current revision
    - keep_none: bool, keep null values
    - sync: bool, wait for sync to disk
    
    Returns:
    Result[Json]: Updated edge metadata
    """
    
def replace_edge(self, edge: Json, check_rev: bool = True,
                 sync=None) -> Result[Json]:
    """
    Replace edge in graph.
    
    Parameters:
    - edge: dict, complete replacement edge
    - check_rev: bool, check current revision
    - sync: bool, wait for sync to disk
    
    Returns:
    Result[Json]: Replaced edge metadata
    """
    
def delete_edge(self, edge, rev=None, check_rev: bool = True,
                ignore_missing: bool = False, sync=None) -> Result:
    """
    Delete edge from graph.
    
    Parameters:
    - edge: str or dict, edge identifier
    - rev: str, expected revision
    - check_rev: bool, check revision matches
    - ignore_missing: bool, ignore if edge doesn't exist
    - sync: bool, wait for sync to disk
    
    Returns:
    Result[Json|bool]: Deletion metadata or boolean
    """

Graph Utilities

Convenience methods for common graph operations and traversals.

def link(self, collection: str, from_vertex, to_vertex, data=None,
         sync=None) -> Result[Json]:
    """
    Create edge between two vertices.
    
    Parameters:
    - collection: str, edge collection name
    - from_vertex: str or dict, source vertex
    - to_vertex: str or dict, target vertex
    - data: dict, additional edge data
    - sync: bool, wait for sync to disk
    
    Returns:
    Result[Json]: Created edge metadata
    """
    
def edges(self, collection: str, vertex, direction=None) -> Result[Json]:
    """
    Get edges connected to vertex.
    
    Parameters:
    - collection: str, edge collection name
    - vertex: str or dict, vertex identifier
    - direction: str, edge direction ('in', 'out', 'any')
    
    Returns:
    Result[Json]: Connected edges information
    """
    
def traverse(self, start_vertex, direction: str = "outbound",
             item_order: str = "forward", strategy=None, **kwargs) -> Result[Json]:
    """
    Traverse graph (deprecated - use AQL instead).
    
    Parameters:
    - start_vertex: str or dict, starting vertex
    - direction: str, traversal direction
    - item_order: str, result ordering
    - strategy: str, traversal strategy
    - **kwargs: additional traversal options
    
    Returns:
    Result[Json]: Traversal results
    """

Usage Examples

Basic Graph Setup

from arango import ArangoClient

client = ArangoClient()
db = client.db('social', username='root', password='password')

# Create a social network graph
edge_definitions = [
    {
        'edge_collection': 'friendships',
        'from_vertex_collections': ['users'],
        'to_vertex_collections': ['users']
    },
    {
        'edge_collection': 'follows',
        'from_vertex_collections': ['users'],
        'to_vertex_collections': ['users']
    }
]

social_graph = db.create_graph(
    'social_network',
    edge_definitions=edge_definitions
)

print(f"Created graph: {social_graph.name}")

Working with Vertices

# Get vertex collection
users = social_graph.vertex_collection('users')

# Insert vertices
alice = users.insert({'name': 'Alice', 'age': 25, 'city': 'Boston'})
bob = users.insert({'name': 'Bob', 'age': 30, 'city': 'NYC'})
charlie = users.insert({'name': 'Charlie', 'age': 28, 'city': 'SF'})

print(f"Created users: {alice['_key']}, {bob['_key']}, {charlie['_key']}")

# Update vertex
users.update({
    '_key': alice['_key'], 
    'occupation': 'Engineer'
})

# Get vertex
user_doc = social_graph.vertex(f"users/{alice['_key']}")
print(f"User: {user_doc['name']}, Age: {user_doc['age']}")

Working with Edges

# Get edge collection
friendships = social_graph.edge_collection('friendships')
follows = social_graph.edge_collection('follows')

# Create friendships (bidirectional)
friendship = friendships.insert({
    '_from': f"users/{alice['_key']}",
    '_to': f"users/{bob['_key']}",
    'since': '2020-01-15',
    'strength': 'close'
})

# Create follow relationships (directional)
follow1 = follows.insert({
    '_from': f"users/{alice['_key']}",
    '_to': f"users/{charlie['_key']}",
    'since': '2021-03-10'
})

follow2 = follows.insert({
    '_from': f"users/{bob['_key']}",
    '_to': f"users/{charlie['_key']}",
    'since': '2021-05-20'
})

# Using convenience link method
social_graph.link(
    'friendships',
    f"users/{bob['_key']}",
    f"users/{charlie['_key']}",
    data={'since': '2021-06-01', 'strength': 'casual'}
)

Edge Definitions Management

# List current edge definitions
definitions = social_graph.edge_definitions()
for definition in definitions:
    print(f"Edge collection: {definition['collection']}")
    print(f"From: {definition['from']}")
    print(f"To: {definition['to']}")

# Add new edge definition
social_graph.create_edge_definition(
    edge_collection='likes',
    from_vertex_collections=['users'],
    to_vertex_collections=['posts']
)

# Create the posts vertex collection
posts_collection = social_graph.create_vertex_collection('posts')

# Update edge definition
social_graph.replace_edge_definition(
    edge_collection='follows',
    from_vertex_collections=['users'],
    to_vertex_collections=['users', 'pages']  # Now can follow pages too
)

Graph Traversal with AQL

# Find friends of friends
query = """
    FOR user IN users
        FILTER user.name == @start_user
        FOR friend IN 1..1 OUTBOUND user friendships
            FOR friend_of_friend IN 1..1 OUTBOUND friend friendships
                FILTER friend_of_friend._key != user._key
                RETURN DISTINCT {
                    name: friend_of_friend.name,
                    city: friend_of_friend.city,
                    mutual_friend: friend.name
                }
"""

cursor = db.aql.execute(query, bind_vars={'start_user': 'Alice'})
friends_of_friends = list(cursor)

# Shortest path between users
shortest_path_query = """
    FOR path IN OUTBOUND SHORTEST_PATH 
        @start_vertex TO @end_vertex 
        GRAPH @graph_name
        RETURN path
"""

cursor = db.aql.execute(
    shortest_path_query,
    bind_vars={
        'start_vertex': f"users/{alice['_key']}",
        'end_vertex': f"users/{charlie['_key']}",
        'graph_name': 'social_network'
    }
)

path_result = cursor.next()
print(f"Shortest path length: {len(path_result['vertices'])}")

Advanced Graph Operations

# Get all edges connected to a vertex
connected_edges = social_graph.edges('friendships', f"users/{alice['_key']}")
print(f"Alice has {len(connected_edges['edges'])} friendship connections")

# Complex traversal with filters
complex_query = """
    FOR user IN users
        FILTER user.city == @city
        FOR connection IN 1..3 ANY user GRAPH @graph_name
            FILTER connection.age BETWEEN @min_age AND @max_age
            COLLECT city = connection.city WITH COUNT INTO city_count
            SORT city_count DESC
            RETURN {
                city: city,
                count: city_count
            }
"""

cursor = db.aql.execute(
    complex_query,
    bind_vars={
        'city': 'Boston',
        'graph_name': 'social_network',
        'min_age': 20,
        'max_age': 40
    }
)

city_stats = list(cursor)
for stat in city_stats:
    print(f"{stat['city']}: {stat['count']} connections")

Graph Analytics

# Centrality analysis
centrality_query = """
    FOR user IN users
        LET connections = (
            FOR edge IN friendships
                FILTER edge._from == user._id OR edge._to == user._id
                RETURN 1
        )
        SORT LENGTH(connections) DESC
        LIMIT 10
        RETURN {
            user: user.name,
            degree: LENGTH(connections)
        }
"""

top_connected = list(db.aql.execute(centrality_query))
print("Most connected users:")
for user in top_connected:
    print(f"{user['user']}: {user['degree']} connections")

# Community detection (simple clustering)
community_query = """
    FOR user IN users
        LET friends = (
            FOR friend IN 1..1 OUTBOUND user friendships
                RETURN friend._key
        )
        LET mutual_connections = (
            FOR friend_key IN friends
                FOR mutual IN 1..1 OUTBOUND CONCAT('users/', friend_key) friendships
                    FILTER mutual._key IN friends
                    RETURN 1
        )
        RETURN {
            user: user.name,
            friends: LENGTH(friends),
            clustering: LENGTH(friends) > 1 ? LENGTH(mutual_connections) / (LENGTH(friends) * (LENGTH(friends) - 1)) : 0
        }
"""

clustering_results = list(db.aql.execute(community_query))
for result in clustering_results:
    print(f"{result['user']}: clustering coefficient = {result['clustering']:.2f}")

Graph Cleanup

# Remove vertices (and connected edges automatically)
social_graph.delete_vertex(f"users/{charlie['_key']}")

# Remove edge definition
social_graph.delete_edge_definition('likes', purge=True)

# Remove vertex collection from graph
social_graph.delete_vertex_collection('posts', purge=True)

# Delete entire graph
db.delete_graph('social_network', drop_collections=False)

Install with Tessl CLI