Accessing Services in Python

The PLANQK Service SDK enables you to access and monitor your deployed services directly from a Python program.

Installation

Begin by installing the PLANQK Service SDK using pip:

pip install --upgrade planqk-service-sdk

Using the SDK

Replace the placeholders your_consumer_key and your_consumer_secret in the code snippet below with the credentials provided in one of your platform applications. Additionally, configure the service_endpoint to the endpoint URL of the subscribed PLANQK service (see list of subscriptions in an application).

# Your consumer key and secret from your application details
consumer_key = "your_consumer_key"
consumer_secret = "your_consumer_secret"

# Service endpoint from the application's subscription details
service_endpoint = "https://gateway.platform.planqk.de/..."

Create a client and start a service execution:

from planqk.service.client import PlanqkServiceClient

# Create a client
client = PlanqkServiceClient(service_endpoint, consumer_key, consumer_secret)

# Prepare your input data and parameters
data = {"input": {"a": 1, "b": 2}}
params = {"param1": "value1", "param2": "value2"}

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

# Wait for the service execution to finish (blocking)
service_execution.wait_for_final_state()

print(f"Service execution finished at '{service_execution.ended_at}' with status '{service_execution.status}'")

You may perform other operations while waiting for the service execution to finish:

while not service_execution.has_finished:
    print("Waiting for service execution to finish...")

Alternatively, you may poll the service execution status:

while service_execution.status not in ["SUCCEEDED", "FAILED", "CANCELLED"]:
    service_execution.refresh()

Use the client to retrieve a service execution by its ID:

service_execution = client.get_service_execution("0030737b-35cb-46a8-88c2-f59d4885484d")

print(f"Service execution finished at '{service_execution.ended_at}' with status '{service_execution.status}'")

Retrieve the result:

result = service_execution.result()

Retrieve service execution logs:

logs = service_execution.logs()

print("...")
for log in logs[-5:]:
    print(log)

List and download the result files:

import os

# List the result files
files = service_execution.result_files()

# Download the result files
cwd = os.getcwd()
for file in files:
    service_execution.download_result_file(file, cwd)

print(files)

Alternatively, you may access the raw byte stream:

file_stream = service_execution.result_file_stream("output.json")

with open("output.json", "wb") as f:
    for chunk in file_stream:
        f.write(chunk)

Use a Data Pool as input data:

from planqk.service.sdk.types.input_data_ref import InputDataRef

# Create a client
client = PlanqkServiceClient(service_endpoint, consumer_key, consumer_secret)

# Create a data pool, and upload a file called 'data.json'. Use the "Copy File Reference" button in 
# the PLANQK Data Pool UI to copy the data pool id, data source descriptor id, and file id. Paste the
# copied values into the file_reference dictionary below.
file_reference = {
    "dataPoolId": "9b943af3-ca78-4d6e-87fd-33129f5330a2",
    "dataSourceDescriptorId": "a1e4f7ab-82d5-4158-98c3-3562635cedd2",
    "fileId": "f9b5c9a2-0101-46a8-9958-383670a2cef4",
}

# Use the InputDataRef class to define a reference to a file in a data pool.
data_ref = InputDataRef.model_validate(
    {
        "data_pool_id": file_reference["dataPoolId"],
        "data_source_descriptor_id": file_reference["dataSourceDescriptorId"],
        "file_id": file_reference["fileId"],
    }
)
params = {"param1": "value1", "param2": "value2"}

# Start a service execution
service_execution = client.run({"dataRef": data_ref, "params": params})

# Wait for the service execution to finish (blocking)
service_execution.wait_for_final_state()

print(f"Service execution finished at '{service_execution.ended_at}' with status '{service_execution.status}'")

Supported Operations

The SDK provides a simple interface to interact with the PLANQK Service API. Use the PlanqkServiceClient class to create a client and start a service execution. The PlanqkServiceExecution class represents a service execution and provides methods to monitor and manage it.

PlanqkServiceClient

Method	Description
__init__	Initializes the client with the service endpoint, consumer key, and secret.
run	Starts a service execution with the provided request data and parameters. Returns a PlanqkServiceExecution object.
get_service_execution	Retrieves a service execution by its ID. Returns a PlanqkServiceExecution object.

PlanqkServiceExecution

Property	Description
id	The ID of the service execution.
status	The last known status of the service execution.
created_at	The creation time of the service execution.
started_at	The start time of the service execution.
ended_at	The end time of the service execution.
has_finished	Checks if the service execution has finished.

Method	Description
wait_for_final_state	Waits for the service execution to reach a final state (blocking).
refresh	Polls the service execution status.
result	Retrieves the result of the service execution.
result_files	Lists the result files of the service execution.
download_result_file	Downloads a specific result file to the specified directory.
result_file_stream	Accesses the raw byte stream of a specific result file.
cancel	Cancels a service execution.
logs	Retrieves the logs of the service execution.