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

Client & Database Management

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

Core client operations for connecting to ArangoDB and database management functionality. The ArangoClient serves as the entry point for all database interactions, while StandardDatabase provides database-level operations.

Capabilities

Client Operations

Initialize and configure the main ArangoDB client with connection settings and database access.

class ArangoClient:
    def __init__(self, hosts='http://127.0.0.1:8529', host_resolver='fallback',
                 resolver_max_tries=None, http_client=None, serializer=None, 
                 deserializer=None, verify_override=None, request_timeout=60,
                 request_compression=None, response_compression=None):
        """
        Initialize ArangoDB client.
        
        Parameters:
        - hosts: str or list, ArangoDB server URLs (default: 'http://127.0.0.1:8529')
        - host_resolver: str, host resolution strategy ('fallback', 'roundrobin', 'random', 'periodic')
        - resolver_max_tries: int, maximum host resolution attempts
        - http_client: HTTPClient, custom HTTP client implementation
        - serializer: callable, JSON serializer function
        - deserializer: callable, JSON deserializer function
        - verify_override: Union[bool, str, None], SSL certificate verification override
        - request_timeout: Union[int, float, None], request timeout in seconds (default: 60)
        - request_compression: RequestCompression, request compression handler
        - response_compression: str, response compression type
        """

    def db(self, name='_system', username='root', password='', verify=False,
           auth_method='basic', user_token=None, superuser_token=None) -> StandardDatabase:
        """
        Connect to database and return StandardDatabase API wrapper.
        
        Parameters:
        - name: str, database name (default '_system')
        - username: str, username for basic authentication
        - password: str, password for basic authentication  
        - verify: bool, verify connection by sending test request
        - auth_method: str, authentication method ('basic', 'jwt')
        - user_token: str, user generated token for user access
        - superuser_token: str, user generated token for superuser access
        
        Returns:
        StandardDatabase: Database interface object
        """

    def close(self) -> None:
        """Close HTTP sessions."""
        
    @property
    def hosts(self) -> Sequence[str]:
        """Return the list of ArangoDB host URLs."""

    @property  
    def version(self) -> str:
        """Return the client version."""

    @property
    def request_timeout(self) -> Any:
        """Return the request timeout of the http client."""

Database Management

Database lifecycle operations including creation, deletion, and listing. These operations are performed through a StandardDatabase instance connected to the _system database.

class StandardDatabase:
    def create_database(self, name: str, users=None, replication_factor=None,
                        write_concern=None, sharding=None) -> Result[bool]:
        """
        Create a new database.
        
        Parameters:
        - name: str, database name
        - users: Sequence[Json], list of users with access to the new database
        - replication_factor: Union[int, str, None], default replication factor for collections
        - write_concern: int, default write concern for collections
        - sharding: str, sharding method for new collections ('flexible', 'single')
        
        Returns:
        Result[bool]: True if database was created successfully
        """

    def delete_database(self, name: str, ignore_missing: bool = False) -> Result[bool]:
        """
        Delete a database.
        
        Parameters:
        - name: str, database name to delete
        - ignore_missing: bool, do not raise exception if database does not exist
        
        Returns:
        Result[bool]: True if database was deleted successfully
        """

    def databases(self) -> Result[List[str]]:
        """
        Return the names of all databases.
        
        Returns:
        Result[List[str]]: List of database names
        """

    def databases_accessible_to_user(self) -> Result[List[str]]:
        """
        Return names of databases accessible by the current user.
        
        Returns:
        Result[List[str]]: List of accessible database names
        """

    def has_database(self, name: str) -> Result[bool]:
        """
        Check if a database exists.
        
        Parameters:
        - name: str, database name
        
        Returns:
        Result[bool]: True if database exists
        """

Server Information

Access server status, version information, statistics, and engine details for monitoring and debugging.

def version(self, details: bool = False) -> Result[Json]:
    """
    Return ArangoDB server version information.
    
    Parameters:
    - details: bool, return detailed version information
    
    Returns:
    Result[Json]: Version information dictionary
    """

def status(self) -> Result[Json]:
    """
    Return ArangoDB server status information.
    
    Returns:
    Result[Json]: Server status dictionary
    """

def statistics(self, description: bool = False) -> Result[Json]:
    """
    Return server statistics.
    
    Parameters:
    - description: bool, return description of each figure
    
    Returns:
    Result[Json]: Server statistics dictionary
    """

def role(self) -> Result[str]:
    """
    Return server role.
    
    Returns:
    Result[str]: Server role ('SINGLE', 'COORDINATOR', 'PRIMARY', 'SECONDARY', 'AGENT', 'UNDEFINED')
    """

def engine(self) -> Result[Json]:
    """
    Return storage engine information.
    
    Returns:
    Result[Json]: Storage engine details dictionary
    """

Cluster Operations

Manage and monitor ArangoDB clusters through the cluster property. These operations are available when connected to a cluster.

@property
def cluster(self) -> Cluster:
    """Return cluster API wrapper."""

class Cluster:
    def server_id(self) -> Result[str]:
        """Return the server ID."""

    def server_role(self) -> Result[str]:
        """Return the server role."""

    def server_version(self, server_id: str) -> Result[str]:
        """Return version of given server."""

    def server_engine(self, server_id: str) -> Result[Json]:
        """Return engine details of given server."""

    def health(self) -> Result[Json]:
        """Return cluster health information."""

    def endpoints(self) -> Result[Json]:
        """Return cluster endpoints information."""

Database Properties

Core database properties and connection information.

@property
def name(self) -> str:
    """Return database name."""
    
@property
def username(self) -> str:
    """Return username."""
    
@property 
def context(self) -> str:
    """Return database context string."""

Usage Examples

Basic Client Setup

from arango import ArangoClient

# Single server
client = ArangoClient(hosts='http://localhost:8529')

# Multiple servers with load balancing
client = ArangoClient(
    hosts=['http://server1:8529', 'http://server2:8529'],
    host_resolver='roundrobin'
)

# Custom timeout and compression
client = ArangoClient(
    hosts='http://localhost:8529',
    request_timeout=120,
    verify_override=False
)

Database Management

# Connect to system database (required for database management)
sys_db = client.db('_system', username='root', password='password')

# Create new database with users
users = [
    {'username': 'admin', 'password': 'secret', 'active': True},
    {'username': 'guest', 'password': 'guest', 'active': True}
]
sys_db.create_database('myapp', users=users)

# Connect to application database
app_db = client.db('myapp', username='admin', password='secret')

# List all databases (requires _system database connection)
db_list = sys_db.databases()
print(db_list)  # ['_system', 'myapp']

# Check if database exists
if sys_db.has_database('myapp'):
    print("Database exists")

# Clean up
sys_db.delete_database('myapp')

Server Monitoring

# Connect to database
db = client.db('_system', username='root', password='password')

# Get server information
version_info = db.version(details=True)
print(f"ArangoDB Version: {version_info['version']}")

server_stats = db.statistics()
print(f"Server uptime: {server_stats.get('uptime', 'N/A')}")

# Check server role
server_role = db.role()
print(f"Server role: {server_role}")

# Engine information
engine_info = db.engine()
print(f"Storage engine: {engine_info['name']}")

Cluster Operations

# Connect to cluster coordinator
db = client.db('_system', username='root', password='password')

# Check if running in cluster mode
if db.role() == 'COORDINATOR':
    cluster = db.cluster()
    
    # Get cluster health
    health = cluster.health()
    print(f"Cluster health: {health}")
    
    # Get cluster endpoints
    endpoints = cluster.endpoints()
    print(f"Cluster endpoints: {endpoints}")
    
    # Get server information
    server_id = cluster.server_id()
    server_role = cluster.server_role()
    print(f"Server ID: {server_id}, Role: {server_role}")

Authentication Methods

# Basic authentication (default)
db = client.db('mydb', username='user', password='pass', auth_method='basic')

# JWT authentication
db = client.db('mydb', username='user', password='pass', auth_method='jwt')

# User token authentication
db = client.db('mydb', user_token='your_jwt_token_here')

# Superuser token authentication
db = client.db('mydb', superuser_token='your_superuser_token_here')

# Verify connection
db = client.db('mydb', username='user', password='pass', verify=True)

Install with Tessl CLI