Service SDK Reference

The Service SDK provides an easy way to interact with Managed Services on the Platform.

This SDK allows you to:

Execute quantum and classical algorithms as managed services
Submit jobs with input data and parameters
Monitor execution status and retrieve results
Work with data pools for input datasets
Access execution logs and download result files

The SDK is available for both Python and TypeScript/JavaScript, making it easy to integrate with your applications.

Installation

You need to have Python 3.9 or higher installed.

Recommended: Using uv

We recommend using uv - a fast Python package and project manager written in Rust. It's significantly faster than pip and provides better dependency resolution.

Install uv

First, install uv if you haven't already:

bash

# On macOS and Linux
curl -LsSf https://astral.sh/uv/install.sh | sh

# On Windows
powershell -c "irm https://astral.sh/uv/install.ps1 | iex"

# Or using pip
pip install uv

Create a new project with uv

bash

# Create a new project directory
mkdir my-planqk-project
cd my-planqk-project

# Initialize a new Python project
uv init

# Add the PLANQK Service SDK
uv add planqk-service-sdk

# Run your Python script
uv run python your_script.py

Add to existing project

If you already have a Python project, you can add the SDK:

bash

# Add the PLANQK Service SDK to your existing project
uv add planqk-service-sdk

# Or add it to a specific dependency group
uv add --group dev planqk-service-sdk  # for development dependencies

Alternative: Using pip

The package is also available on PyPI and can be installed via pip:

bash

# Create a virtual environment (recommended)
python -m venv planqk-env
source planqk-env/bin/activate  # On Windows: planqk-env\Scripts\activate

# Install the package
pip install planqk-service-sdk

TIP

Whether using uv or pip, it's recommended to use a virtual environment to install the SDK in isolation and avoid dependency conflicts.

After installing the SDK, you can start using it to interact with PLANQK Managed Services:

Follow the Authentication section to set up your credentials
Follow the Using the Service SDK section to begin executing services and managing jobs

Authentication

To use the SDK, you need to authenticate using consumer credentials (consumer key and consumer secret). You can generate these credentials for your services on the PLANQK Platform in your Service Settings.

Consumer credentials can be set in two ways:

Environment Variables: Set the following environment variables:

bash

export CONSUMER_KEY="your_consumer_key"
export CONSUMER_SECRET="your_consumer_secret"
export SERVICE_ENDPOINT="https://your-service-endpoint.planqk.de"
export TOKEN_ENDPOINT="https://gateway.platform.planqk.de/token"  # optional, uses default if not set

Explicitly during instantiation: Pass the credentials directly to the PlanqkServiceClient:

python

from planqk.service.client import PlanqkServiceClient

client = PlanqkServiceClient(
    service_endpoint="https://your-service-endpoint.planqk.de",
    consumer_key="your_consumer_key",
    consumer_secret="your_consumer_secret",
    token_endpoint="https://gateway.platform.planqk.de/token"  # optional
)

TIP

When using environment variables in Python applications, consider using the python-dotenv package to load variables from a .env file:

bash

# Using uv
uv add python-dotenv

# Using pip
pip install python-dotenv

Then in your application:

python

import os
from dotenv import load_dotenv
from planqk.service.client import PlanqkServiceClient

load_dotenv()  # Load variables from .env file

client = PlanqkServiceClient(
    service_endpoint=os.getenv("SERVICE_ENDPOINT"),
    consumer_key=os.getenv("CONSUMER_KEY"),
    consumer_secret=os.getenv("CONSUMER_SECRET"),
    token_endpoint=os.getenv("TOKEN_ENDPOINT")  # optional
)

Using the Service SDK

The Service SDK provides a simple interface to execute managed services and handle their results.

Basic Usage

Here's a basic example of how to use the Service SDK:

python

from planqk.service.client import PlanqkServiceClient

# Create a client instance
client = PlanqkServiceClient(
    service_endpoint="https://your-service-endpoint.planqk.de",
    consumer_key="your_consumer_key",
    consumer_secret="your_consumer_secret"
)

# Prepare your input data and parameters
data = {"values": [1, 2, 3, 4, 5]}
params = {"algorithm": "quantum_fourier_transform", "shots": 1000}

# Start a service execution
execution = client.run(request={"data": data, "params": params})

# Wait for completion and get results
execution.wait_for_final_state()
result = execution.result()

print(f"Execution finished with status: {execution.status}")
print(f"Result: {result}")

Working with Service Executions

The PlanqkServiceExecution class provides methods to monitor and interact with running services:

Monitoring Execution Status

python

# Check if execution has finished
if execution.has_finished:
    print("Execution completed!")
else:
    print(f"Current status: {execution.status}")

# Wait for completion with timeout
try:
    execution.wait_for_final_state(timeout=300)  # 5 minutes timeout
    print("Execution completed successfully!")
except TimeoutError:
    print("Execution timed out")

# Manual polling with custom logic
while not execution.has_finished:
    execution.refresh()  # Update status
    print(f"Status: {execution.status}")
    time.sleep(10)  # Wait 10 seconds before next check

Execution Properties

python

# Access execution properties
print(f"Execution ID: {execution.id}")
print(f"Status: {execution.status}")
print(f"Created at: {execution.created_at}")
print(f"Started at: {execution.started_at}")
print(f"Ended at: {execution.ended_at}")

Retrieving Results

Once an execution completes successfully, you can access its results:

python

# Get the complete result object
result = execution.result()
print(result)

# Access execution logs
logs = execution.logs()
print("Recent logs:")
for log in logs[-5:]:  # Print last 5 log entries
    print(log)

Working with Result Files

Many services produce output files that you can download:

python

# List available result files
files = execution.result_files()
print(f"Available files: {files}")

# Download files to a specific directory
import os
download_dir = "/path/to/download/directory"

for file_name in files:
    execution.download_result_file(file_name, download_dir)
    print(f"Downloaded: {file_name}")

# Access raw file stream for custom processing
file_stream = execution.result_file_stream("output.json")
with open("local_output.json", "wb") as f:
    for chunk in file_stream:
        f.write(chunk)

Using Data Pools

Data pools allow you to reference datasets stored on the PLANQK Platform:

python

from planqk.service.client import PlanqkServiceClient
from planqk.service.datapool import DataPoolReference

client = PlanqkServiceClient(
    service_endpoint="https://your-service-endpoint.planqk.de",
    consumer_key="your_consumer_key",
    consumer_secret="your_consumer_secret"
)

# Reference a data pool by ID
dataset = DataPoolReference(id="89ae42e1-9697-4742-adfb-95bf018104c3")

# Use the data pool in your service execution
data = {"input_file": "dataset.csv"}
execution = client.run(request={"data": data, "dataset": dataset})

execution.wait_for_final_state()
print(f"Execution completed with status: {execution.status}")

Managing Service Executions

You can retrieve and manage existing service executions:

python

# Get a specific execution by ID
execution = client.get_service_execution("3063bd47-1b73-414f-96da-95a95d8fdb0c")

# List all your service executions
executions = client.get_service_executions()
print(f"Found {len(executions)} executions")

for exec in executions:
    print(f"ID: {exec.id}, Status: {exec.status}, Created: {exec.created_at}")

# Cancel a running execution
if not execution.has_finished:
    execution.cancel()
    print("Execution cancelled")

Error Handling

The SDK provides proper error handling for common scenarios:

python

from planqk.service.client import PlanqkServiceClient

try:
    client = PlanqkServiceClient(
        service_endpoint="https://your-service-endpoint.planqk.de",
        consumer_key="invalid_key",
        consumer_secret="invalid_secret"
    )
    
    execution = client.run(request={"data": {}, "params": {}})
    execution.wait_for_final_state()
    
    if execution.status == "FAILED":
        logs = execution.logs()
        print("Execution failed. Recent logs:")
        for log in logs[-10:]:
            print(log)
    else:
        result = execution.result()
        print(f"Success: {result}")
        
except Exception as e:
    print(f"Error: {e}")

Advanced Usage Examples

Batch Processing

python

import time
from concurrent.futures import ThreadPoolExecutor

def process_batch(batch_data):
    """Process a batch of data"""
    execution = client.run(request={"data": batch_data, "params": {"mode": "batch"}})
    execution.wait_for_final_state()
    return execution.result()

# Process multiple batches concurrently
batches = [
    {"values": list(range(0, 100))},
    {"values": list(range(100, 200))},
    {"values": list(range(200, 300))}
]

with ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(process_batch, batches))

print(f"Processed {len(results)} batches")

Custom Polling Strategy

python

def wait_with_progress(execution, timeout=None):
    """Wait for execution with progress updates"""
    start_time = time.time()
    last_status = None
    
    while not execution.has_finished:
        execution.refresh()
        
        if execution.status != last_status:
            elapsed = time.time() - start_time
            print(f"[{elapsed:.1f}s] Status changed to: {execution.status}")
            last_status = execution.status
        
        if timeout and (time.time() - start_time) > timeout:
            raise TimeoutError("Execution timed out")
            
        time.sleep(5)
    
    print(f"Execution completed with status: {execution.status}")

# Usage
execution = client.run(request={"data": data, "params": params})
wait_with_progress(execution, timeout=300)

API Reference

PlanqkServiceClient

The main client class for interacting with PLANQK Managed Services.

Constructor

python

PlanqkServiceClient(
    service_endpoint: str,
    consumer_key: Optional[str],
    consumer_secret: Optional[str],
    token_endpoint: str = "https://gateway.platform.planqk.de/token"
)

Methods

Method	Description
`run(request: Dict)` → `PlanqkServiceExecution`	Start a new service execution with the provided request data
`get_service_execution(id: str)` → `PlanqkServiceExecution`	Retrieve a specific service execution by ID
`get_service_executions()` → `List[PlanqkServiceExecution]`	Retrieve all service executions for the current user

ServiceExecution

Represents a single service execution and provides methods to monitor and interact with it.

Properties

Property	Type	Description
`id`	`str`	Unique identifier of the execution
`status`	`ServiceExecutionStatus`	Current status of the execution
`created_at`	`str`	Timestamp when the execution was created
`started_at`	`Optional[str]`	Timestamp when the execution started
`ended_at`	`Optional[str]`	Timestamp when the execution ended
`has_finished`	`bool`	Whether the execution has reached a final state

Methods

Method	Description
`wait_for_final_state(timeout: Optional[float], wait: float = 5)`	Wait until execution reaches a final state
`refresh()`	Update the execution status from the server
`result()` → `GetResultResponse`	Get the execution result (blocks until completion)
`result_files()` → `List[str]`	List names of available result files
`result_file_stream(file_name: str)` → `Iterator[bytes]`	Get raw byte stream for a result file
`download_result_file(file_name: str, target_path: str)`	Download a result file to the specified directory
`cancel()`	Cancel the execution
`logs()` → `List[str]`	Retrieve execution logs

DataPoolReference

Reference to a dataset stored in a PLANQK data pool.

python

DataPoolReference(id: str)

Status Values

Service executions can have the following status values:

UNKNOWN: Status is unknown or not yet determined
PENDING: Execution is pending and waiting to be processed
RUNNING: Execution is currently running
SUCCEEDED: Execution completed successfully
CANCELLED: Execution was cancelled by the user
FAILED: Execution failed

Best Practices

Resource Management: Always ensure executions complete or are explicitly cancelled to avoid resource leaks
Error Handling: Check execution status and handle failures appropriately
Timeouts: Use reasonable timeouts to avoid indefinite waiting
Logging: Monitor execution logs for debugging and troubleshooting
File Management: Clean up downloaded result files when no longer needed

python

# Example of good resource management
try:
    execution = client.run(request=request_data)
    execution.wait_for_final_state(timeout=600)  # 10 minute timeout
    
    if execution.status == "SUCCEEDED":
        result = execution.result()
        # Process result...
    else:
        logs = execution.logs()
        print(f"Execution failed. Logs: {logs[-5:]}")
        
except TimeoutError:
    print("Execution timed out, cancelling...")
    execution.cancel()
except Exception as e:
    print(f"Error occurred: {e}")
    if 'execution' in locals() and not execution.has_finished:
        execution.cancel()

Service SDK Reference ​

Installation ​

Recommended: Using uv ​

Install uv ​

Create a new project with uv ​

Add to existing project ​

Alternative: Using pip ​

Authentication ​

Using the Service SDK ​

Basic Usage ​

Working with Service Executions ​

Monitoring Execution Status ​

Execution Properties ​

Retrieving Results ​

Working with Result Files ​

Using Data Pools ​

Managing Service Executions ​

Error Handling ​

Advanced Usage Examples ​

Batch Processing ​

Custom Polling Strategy ​

API Reference ​

PlanqkServiceClient ​

Constructor ​

Methods ​

ServiceExecution ​

Properties ​

Methods ​

DataPoolReference ​

Status Values ​

Best Practices ​

Service SDK Reference

Installation

Recommended: Using uv

Install uv

Create a new project with uv

Add to existing project

Alternative: Using pip

Authentication

Using the Service SDK

Basic Usage

Working with Service Executions

Monitoring Execution Status

Execution Properties

Retrieving Results

Working with Result Files

Using Data Pools

Managing Service Executions

Error Handling

Advanced Usage Examples

Batch Processing

Custom Polling Strategy

API Reference

PlanqkServiceClient

Constructor

Methods

ServiceExecution

Properties

Methods

DataPoolReference

Status Values

Best Practices