Python Integration

This guide demonstrates how to use Python to interact with the Surflux gRPC API. Python's simplicity and rich ecosystem make it ideal for data analysis, scripting, and rapid prototyping.

Overview

Python provides excellent gRPC support through the grpcio library. This guide walks you through setting up a complete Python project with proper authentication and proto code generation.

Prerequisites

• Python 3.8 or later
• pip (Python package manager)

Step 1: Create a New Project Directory

Create a dedicated directory for your Sui gRPC project:

mkdir sui-grpc-python
cd sui-grpc-python

Step 2: Create Virtual Environment (Recommended)

Create and activate a Python virtual environment to isolate dependencies:

python3 -m venv venv

# Activate on macOS/Linux
source venv/bin/activate

# Activate on Windows
venv\Scripts\activate

Step 3: Install gRPC and Protobuf Dependencies

Install the core gRPC and Protocol Buffer libraries:

pip install grpcio grpcio-tools protobuf googleapis-common-protos

Step 4: Organize Your Project Directory

Create a protos folder to store the Protocol Buffer definition files:

mkdir protos

Download the official Sui proto files from the MystenLabs/sui-apis repository. Clone the repository and copy the proto files:

git clone https://github.com/MystenLabs/sui-apis.git protos/sui-apis

Your project structure should look like this:

sui-grpc-python/
├── venv/
├── protos/
│   └── sui-apis/
│       └── proto/
│           └── sui/
│               └── rpc/
│                   └── v2/
│                       ├── ledger_service.proto
│                       ├── common.proto
│                       └── ... (other proto files)
└── client.py

Step 5: Generate Python Code from Proto Files

Generate the _pb2.py and _pb2_grpc.py files by running:

python -m grpc_tools.protoc \
  -I./protos/sui-apis/proto \
  --python_out=. \
  --grpc_python_out=. \
  ./protos/sui-apis/proto/sui/rpc/v2/*.proto

# Create __init__.py files to make the generated code a proper package
touch sui/__init__.py
touch sui/rpc/__init__.py
touch sui/rpc/v2/__init__.py

This will generate all the required Python files for the Sui gRPC services.

Step 6: Create Client Implementation

Create a client.py file with the following helper function for authentication:

import grpc
from sui.rpc.v2 import ledger_service_pb2_grpc

SERVER_ADDR = 'grpc.surflux.dev:443'

def create_sui_grpc_client(api_key: str):
    """
    Create a gRPC client with Surflux API key authentication.
    
    Args:
        api_key: Your Surflux API key
        
    Returns:
        tuple: (stub, metadata) where stub is the service client and metadata contains auth
    """
    # Create secure SSL channel
    credentials = grpc.ssl_channel_credentials()
    channel = grpc.secure_channel(SERVER_ADDR, credentials)
    
    # Create service stub for Ledger operations
    stub = ledger_service_pb2_grpc.LedgerServiceStub(channel)
    
    # Prepare authentication metadata with API key
    metadata = [('x-api-key', api_key)]
    
    return stub, metadata

Complete Example

Here's a complete example showing how to fetch epoch information:

import grpc
from sui.rpc.v2 import ledger_service_pb2, ledger_service_pb2_grpc

SERVER_ADDR = 'grpc.surflux.dev:443'
API_KEY = 'YOUR_API_KEY'

def create_sui_grpc_client(api_key: str):
    """Create an authenticated gRPC client for Surflux."""
    credentials = grpc.ssl_channel_credentials()
    channel = grpc.secure_channel(SERVER_ADDR, credentials)
    stub = ledger_service_pb2_grpc.LedgerServiceStub(channel)
    metadata = [('x-api-key', api_key)]
    return stub, metadata

def get_current_epoch():
    """Fetch current epoch information from Surflux."""
    # Create client
    stub, metadata = create_sui_grpc_client(API_KEY)
    
    # Create request
    request = ledger_service_pb2.GetEpochRequest()
    
    # Make the call with metadata
    response = stub.GetEpoch(request, metadata=metadata)
    
    print(f"Current Epoch: {response.epoch.epoch}")
    print(f"Epoch Start Timestamp: {response.epoch.start}")
    return response

if __name__ == '__main__':
    get_current_epoch()

Making Different Requests

Once your proto code is generated, you can make requests to any service:

Example: Get a Checkpoint

def get_checkpoint(sequence_number: int):
    """Fetch a specific checkpoint by sequence number."""
    stub, metadata = create_sui_grpc_client(API_KEY)
    
    request = ledger_service_pb2.GetCheckpointRequest(
        sequence_number=sequence_number
    )
    
    try:
        response = stub.GetCheckpoint(request, metadata=metadata)
        print(f"Checkpoint Digest: {response.checkpoint.digest}")
        print(f"Timestamp: {response.checkpoint.summary.timestamp}")
        return response
    except grpc.RpcError as e:
        print(f"Error: {e.code()} - {e.details()}")
        return None

# Usage
get_checkpoint(1000)

Example: Get Object

from sui.rpc.v2 import state_service_pb2, state_service_pb2_grpc

def get_object(object_id: str):
    """Fetch an object by its ID."""
    credentials = grpc.ssl_channel_credentials()
    channel = grpc.secure_channel(SERVER_ADDR, credentials)
    stub = state_service_pb2_grpc.StateServiceStub(channel)
    metadata = [('x-api-key', API_KEY)]
    
    request = state_service_pb2.GetObjectRequest(
        object_id=object_id
    )
    
    try:
        response = stub.GetObject(request, metadata=metadata)
        print(f"Object Version: {response.object.version}")
        print(f"Owner: {response.object.owner}")
        return response
    except grpc.RpcError as e:
        print(f"Error: {e.code()} - {e.details()}")
        return None

Best Practices

1. Use Virtual Environments: Always isolate your project dependencies.
2. Error Handling: Wrap gRPC calls in try-except blocks to handle RpcError.
3. Timeouts: Set appropriate timeouts for requests using the timeout parameter.
4. Connection Reuse: Reuse channels and stubs instead of creating new ones for each request.
5. Type Hints: Use Python type hints for better code maintainability.

Handling Errors

try:
    response = stub.GetCheckpoint(request, metadata=metadata, timeout=5.0)
except grpc.RpcError as e:
    if e.code() == grpc.StatusCode.UNAUTHENTICATED:
        print("Invalid API key")
    elif e.code() == grpc.StatusCode.NOT_FOUND:
        print("Checkpoint not found")
    elif e.code() == grpc.StatusCode.DEADLINE_EXCEEDED:
        print("Request timed out")
    else:
        print(f"Unknown error: {e.details()}")

Next Steps

• Explore the generated proto definitions to see all available services and methods
• Implement async gRPC clients using grpc.aio for better performance
• Add logging and monitoring to your production applications