REST API

from processcube_client.core.api import Client
 
# Client erstellen
client = Client('http://localhost:8000')
 
# Mit Authentifizierung
client = Client(
    'http://localhost:8000',
    client_id='my-service',
    client_secret='secret-key',
    authority_url='http://localhost:11235'
)

# Application Info
info = client.info()
 
print(f"Engine Version: {info['version']}")
print(f"Name: {info['name']}")

# Authority Info abrufen
authority_info = client.authority_info()
 
print(f"Authority URL: {authority_info['url']}")
print(f"Authority Version: {authority_info['version']}")

# Prozess aus Datei deployen
with open('OrderProcess.bpmn', 'r') as file:
    bpmn_xml = file.read()
 
response = client.process_definition_deploy(
    process_model_id='OrderProcess',
    xml=bpmn_xml
)
 
print(f"Deployed: {response['processDefinitionId']}")

# Direktes Deployment über Pfad
response = client.process_definition_deploy_via_pathname(
    'path/to/OrderProcess.bpmn'
)
 
print(f"Deployed: {response['processDefinitionId']}")

# Alle Prozess-Modelle
process_models = client.get_process_models()
 
for model in process_models:
    print(f"Prozess: {model['id']} - {model['name']}")

from processcube_client.core.api import ProcessStartRequest, StartCallbackType
 
# Start-Request erstellen
request = ProcessStartRequest(
    return_on=StartCallbackType.CallbackOnProcessInstanceCreated,
    process_model_id='OrderProcess',
    initial_token={
        'customerName': 'Max Mustermann',
        'orderAmount': 1500
    }
)
 
# Prozess starten
response = client.process_model_start('OrderProcess', request)
 
print(f"Prozess-Instanz: {response.process_instance_id}")
print(f"Correlation-ID: {response.correlation_id}")

# Start und warten auf Prozess-Ende
request = ProcessStartRequest(
    return_on=StartCallbackType.CallbackOnProcessInstanceFinished,
    process_model_id='OrderProcess',
    initial_token={'orderAmount': 500}
)
 
response = client.process_model_start('OrderProcess', request)
 
print(f"Prozess beendet mit Result: {response.result}")

# Query für Prozess-Instanzen
instances = client.process_instances_query({
    'process_model_id': 'OrderProcess',
    'state': 'running'
})
 
for instance in instances:
    print(f"Instanz: {instance.process_instance_id}")
    print(f"Status: {instance.state}")
    print(f"Gestartet: {instance.created_at}")

# Prozess terminieren
process_instance_id = 'abc-123-def'
 
client.process_instance_terminate(process_instance_id)
 
print(f"Prozess {process_instance_id} terminiert")

# Alle FlowNodes einer Prozess-Instanz
process_instance_id = 'abc-123-def'
 
flow_nodes = client.flow_node_instances_get(process_instance_id)
 
for node in flow_nodes:
    print(f"FlowNode: {node['flowNodeId']}")
    print(f"Type: {node['flowNodeType']}")
    print(f"Status: {node['state']}")

# Signal an alle wartenden Prozesse
client.events_trigger_signal('OrderApproved', {
    'orderId': '12345',
    'approvedBy': 'manager@example.com'
})
 
print("Signal gesendet")

# Message an spezifische Prozess-Instanz
correlation_id = 'order-12345'
 
client.events_trigger_message(
    'PaymentReceived',
    correlation_id,
    {
        'amount': 1500,
        'transactionId': 'TXN-67890'
    }
)
 
print("Message gesendet")

# Alle verfügbaren User Tasks
user_tasks = client.user_tasks_get()
 
for task in user_tasks:
    print(f"Task: {task['userTaskInstanceId']}")
    print(f"Name: {task['flowNodeId']}")
    print(f"Token: {task['token']}")

# Task für Benutzer reservieren
user_task_id = 'task-123'
owner_id = 'user@example.com'
 
client.user_tasks_reserve(user_task_id, owner_id)
 
print(f"Task {user_task_id} reserviert für {owner_id}")

# Reservierung aufheben
user_task_id = 'task-123'
 
client.user_tasks_cancel_reservation(user_task_id)
 
print("Reservierung aufgehoben")

# Task mit Result abschließen
user_task_id = 'task-123'
 
result = {
    'approved': True,
    'comment': 'Genehmigt am 2024-01-15'
}
 
client.user_tasks_finish(user_task_id, result)
 
print("User Task abgeschlossen")

# Fetch and Lock
external_tasks = client.external_task_fetch_and_lock(
    topic='ProcessPayment',
    max_tasks=10,
    lock_duration=60000,  # 60 Sekunden
    worker_id='payment-worker'
)
 
for task in external_tasks:
    print(f"External Task: {task['id']}")
    print(f"Topic: {task['topic']}")

# Task als erfolgreich markieren
task_id = 'ext-task-123'
 
result = {
    'paymentStatus': 'success',
    'transactionId': 'TXN-12345'
}
 
client.external_task_finish(task_id, result)
 
print("External Task abgeschlossen")

# Alle Manual Tasks
manual_tasks = client.manual_tasks_get()
 
for task in manual_tasks:
    print(f"Manual Task: {task['id']}")

# Manual Task beenden
manual_task_id = 'manual-123'
 
client.manual_tasks_finish(manual_task_id, {})
 
print("Manual Task abgeschlossen")

# Empty Tasks abrufen
empty_tasks = client.empty_tasks_get()
 
# Ersten Task abschließen
if empty_tasks:
    task_id = empty_tasks[0]['id']
    client.empty_tasks_finish(task_id)
    print(f"Empty Task {task_id} abgeschlossen")

# DataObjects einer Prozess-Instanz
process_instance_id = 'abc-123-def'
 
data_objects = client.data_object_instances_get(process_instance_id)
 
for obj in data_objects:
    print(f"DataObject: {obj['id']}")
    print(f"Value: {obj['value']}")

from processcube_client.exceptions import ProcessCubeException
 
try:
    response = client.process_model_start('OrderProcess', request)
    print(f"Erfolg: {response.process_instance_id}")
 
except ProcessCubeException as e:
    print(f"Engine-Fehler: {e}")
 
except Exception as e:
    print(f"Allgemeiner Fehler: {e}")

from typing import Dict, List, Any
from processcube_client.core.api import Client
 
def get_running_processes(
    client: Client,
    process_model_id: str
) -> List[Dict[str, Any]]:
    """Gibt alle laufenden Prozess-Instanzen zurück."""
 
    return client.process_instances_query({
        'process_model_id': process_model_id,
        'state': 'running'
    })

import os
from processcube_client.core.api import Client
 
class ProcessCubeService:
    """Service-Klasse für ProcessCube-Operationen."""
 
    def __init__(self):
        self.client = Client(
            os.getenv('ENGINE_URL', 'http://localhost:8000'),
            client_id=os.getenv('CLIENT_ID'),
            client_secret=os.getenv('CLIENT_SECRET'),
            authority_url=os.getenv('AUTHORITY_URL')
        )
 
    def start_order_process(self, customer_name: str, amount: float):
        request = ProcessStartRequest(
            return_on=StartCallbackType.CallbackOnProcessInstanceCreated,
            process_model_id='OrderProcess',
            initial_token={
                'customerName': customer_name,
                'orderAmount': amount
            }
        )
 
        return self.client.process_model_start('OrderProcess', request)
 
# Verwendung
service = ProcessCubeService()
response = service.start_order_process('Max Mustermann', 1500)