Skip to Content
DocsClientsPythonPython Client

Python Client

Der ProcessCube® Python Client ist eine vollständige Python-Library für die Integration mit der ProcessCube® Engine. Sie bietet eine typisierte, pythonische API mit Unterstützung für synchrone und asynchrone Operationen.

Überblick

Der Python Client ermöglicht die nahtlose Integration von ProcessCube® Prozessen in Python-Anwendungen. Mit Type Hints, modernem async/await-Support und einer intuitiven API ist er ideal für Backend-Services, Data Science-Workflows und Automatisierungs-Scripts.

Hauptfeatures

  • Type Hints: Vollständige Typ-Annotationen für bessere IDE-Unterstützung
  • Sync & Async: Sowohl synchrone als auch asynchrone API verfügbar
  • External Task Pattern: Einfache Worker-Implementation
  • User Task Handling: Vollständige User-Task-Unterstützung
  • OAuth 2.0: Integration mit ProcessCube® Authority
  • Pythonic API: Idiomatisches Python-Design
  • Logging: Integriertes Logging mit Python’s logging-Modul

Installation

pip installieren

pip install processcube-client

Mit Poetry

poetry add processcube-client

Mit pip und Requirements

# requirements.txt
processcube-client>=1.0.0
pip install -r requirements.txt

Der Python Client benötigt Python 3.8 oder höher.

Quick Start

Client erstellen

Der einfachste Weg, einen Client zu erstellen:

from processcube_client import Client
 
# Client ohne Authentifizierung (nur für lokale Entwicklung)
client = Client('http://localhost:8000')

Mit Authentifizierung

Für produktive Umgebungen mit ProcessCube® Authority:

from processcube_client import Client
 
client = Client(
    'http://localhost:8000',
    client_id='your-client-id',
    client_secret='your-client-secret',
    authority_url='http://localhost:11235'
)

Prozess starten

Ein einfaches Beispiel zum Starten eines Prozesses:

# Prozess ohne Parameter starten
result = client.start_process('MyProcessKey')
 
print(f'Process started: {result.process_instance_id}')

Mit Parametern starten

# Prozess mit Eingabeparametern starten
input_values = {
    'customer_name': 'Max Mustermann',
    'order_amount': 1500
}
 
result = client.start_process('OrderProcess', input_values)
 
print(f'Process Instance ID: {result.process_instance_id}')
print(f'Result: {result.result}')

Wichtige Konzepte

Prozess-Management

Application Info abrufen

# Engine-Informationen abrufen
info = client.info()
 
print(f'Engine Version: {info.version}')
print(f'Engine URL: {info.url}')

Prozesse abfragen

# Alle deployten Prozesse abrufen
processes = client.get_processes()
 
for process in processes:
    print(f'{process.key}: {process.name}')

Prozess-Instanzen abfragen

from processcube_client import ProcessInstanceQuery
 
# Query erstellen
query = ProcessInstanceQuery(
    process_key='OrderProcess',
    state='running'
)
 
# Prozess-Instanzen abrufen
instances = client.get_process_instances(query)
 
for instance in instances:
    print(f'Instance: {instance.process_instance_id}')
    print(f'State: {instance.state}')

User Task Handling

User Tasks abrufen

from processcube_client import UserTaskQuery
 
# Alle offenen User Tasks abrufen
user_tasks = client.user_task_get()
 
# Mit Query filtern
query = UserTaskQuery(process_instance_id='instance-123')
tasks = client.user_task_get(query)
 
for task in tasks:
    print(f'Task: {task.id}')
    print(f'Form: {task.user_task_config.custom_form}')
    print(f'Fields: {task.user_task_config.form_fields}')

User Task abschließen

# User Task mit Ergebnis abschließen
result_data = {
    'approved': True,
    'comment': 'Genehmigt durch Manager'
}
 
client.user_task_finish(task.id, result_data)

External Task Pattern

Der Python Client bietet einen einfachen External Task Worker:

from processcube_client.external_task import ExternalTaskClient
import logging
 
# Logger konfigurieren
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
 
# Handler-Funktion definieren
def email_handler(payload):
    logger.info(f"Sending email with payload: {payload}")
 
    # E-Mail-Logik hier
    recipient = payload.get('recipient')
    subject = payload.get('subject')
    body = payload.get('body')
 
    # Pseudo: E-Mail versenden
    send_email(recipient, subject, body)
 
    logger.info("Email sent successfully")
 
    # Ergebnis zurückgeben
    return {'success': True, 'sent_at': datetime.now().isoformat()}
 
# External Task Client erstellen
engine_url = 'http://localhost:8000'
et_client = ExternalTaskClient(engine_url)
 
# Task-Topic abonnieren
et_client.subscribe_to_external_task_for_topic(
    'SendEmail',
    email_handler,
    max_tasks=5
)
 
# Worker starten (blockierend)
et_client.start()

Der External Task Worker pollt die Engine automatisch nach neuen Tasks und führt die Handler-Funktion aus.

Event Handling

Signal senden

# Signal an laufende Prozesse senden
client.send_signal(
    'PaymentReceived',
    payload={
        'order_id': 'ORDER-123',
        'amount': 1500
    }
)

Message senden

# Message an spezifische Prozess-Instanz senden
client.send_message(
    'UpdateOrder',
    payload={
        'status': 'shipped',
        'tracking_number': 'TRACK-456'
    },
    process_instance_id='instance-123'
)

Authentifizierung

Client Credentials Flow

Für Service-to-Service Kommunikation:

from processcube_client import Client
 
client = Client(
    'http://localhost:8000',
    client_id='my-service',
    client_secret='secret-key',
    authority_url='http://localhost:11235'
)
 
# Client authentifiziert sich automatisch beim ersten Request
processes = client.get_processes()

Mit Identity Claims

# Client mit spezifischen Identity Claims erstellen
client = Client(
    'http://localhost:8000',
    client_id='my-service',
    client_secret='secret-key',
    authority_url='http://localhost:11235',
    identity_claims={
        'sub': 'user-123',
        'name': 'Max Mustermann',
        'email': 'max@example.com'
    }
)

Type Hints und IDE-Support

Der Python Client nutzt Type Hints für bessere IDE-Unterstützung:

from processcube_client import Client, ProcessInstance, UserTask
from typing import Dict, Any
 
def start_order(client: Client, order_data: Dict[str, Any]) -> ProcessInstance:
    """Startet einen Bestellprozess."""
    result: ProcessInstance = client.start_process('OrderProcess', order_data)
    return result
 
def handle_user_task(client: Client, task: UserTask) -> None:
    """Verarbeitet einen User Task."""
    form_data: Dict[str, Any] = {
        'approved': True,
        'comment': 'OK'
    }
    client.user_task_finish(task.id, form_data)

Best Practices

Error Handling

from processcube_client import ProcessCubeError, ProcessNotFoundError
 
try:
    result = client.start_process('NonExistentProcess')
except ProcessNotFoundError as e:
    print(f'Process does not exist: {e.process_key}')
except ProcessCubeError as e:
    print(f'Engine error: {e.message}')
except Exception as e:
    print(f'Unknown error: {e}')

Connection Pooling

# engine_client.py
from processcube_client import Client
 
# Globaler Client für die gesamte Anwendung
engine_client = Client('http://localhost:8000')
 
# In anderen Modulen verwenden
from engine_client import engine_client
 
result = engine_client.start_process('MyProcess')

Environment Configuration

import os
from processcube_client import Client
 
# Konfiguration aus Umgebungsvariablen
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', 'http://localhost:11235')
)

Logging

import logging
from processcube_client import Client
 
# Logging konfigurieren
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
 
# Client-Logging ist automatisch aktiviert
client = Client('http://localhost:8000')

Framework-Integration

FastAPI

from fastapi import FastAPI, HTTPException
from processcube_client import Client
from pydantic import BaseModel
 
app = FastAPI()
client = Client('http://localhost:8000')
 
class OrderRequest(BaseModel):
    customer_name: str
    order_amount: float
 
@app.post('/start-process')
async def start_process(order: OrderRequest):
    try:
        result = client.start_process('OrderProcess', order.dict())
        return {
            'success': True,
            'process_instance_id': result.process_instance_id
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Flask

from flask import Flask, request, jsonify
from processcube_client import Client
 
app = Flask(__name__)
client = Client('http://localhost:8000')
 
@app.route('/start-process', methods=['POST'])
def start_process():
    try:
        data = request.json
        result = client.start_process('OrderProcess', data)
        return jsonify({
            'success': True,
            'process_instance_id': result.process_instance_id
        })
    except Exception as e:
        return jsonify({'error': str(e)}), 500
 
if __name__ == '__main__':
    app.run(port=5000)

Django

from django.http import JsonResponse
from django.views import View
from processcube_client import Client
import json
 
client = Client('http://localhost:8000')
 
class StartProcessView(View):
    def post(self, request):
        try:
            data = json.loads(request.body)
            result = client.start_process('OrderProcess', data)
            return JsonResponse({
                'success': True,
                'process_instance_id': result.process_instance_id
            })
        except Exception as e:
            return JsonResponse({'error': str(e)}, status=500)

Celery Task

from celery import Celery
from processcube_client import Client
 
app = Celery('tasks', broker='redis://localhost:6379')
client = Client('http://localhost:8000')
 
@app.task
def process_order(order_data):
    """Celery Task zum Starten eines Prozesses."""
    result = client.start_process('OrderProcess', order_data)
    return result.process_instance_id

Troubleshooting

Connection Issues

Problem: Client kann keine Verbindung zur Engine herstellen

Lösungen:

  • Prüfen Sie, ob die Engine-URL korrekt ist
  • Stellen Sie sicher, dass die Engine läuft
  • Überprüfen Sie Firewall-Einstellungen
  • Testen Sie mit: curl http://localhost:8000/api/info

Authentication Errors

Problem: 401 Unauthorized Fehler

Lösungen:

  • Prüfen Sie client_id und client_secret
  • Stellen Sie sicher, dass der Client in der Authority registriert ist
  • Überprüfen Sie die authority_url
  • Prüfen Sie Scopes in der Authority-Konfiguration

Import Errors

Problem: ModuleNotFoundError: No module named 'processcube_client'

Lösungen:

  • Stellen Sie sicher, dass das Paket installiert ist: pip install processcube-client
  • Verwenden Sie ein Virtual Environment
  • Überprüfen Sie die Python-Version (>= 3.8 erforderlich)

Weitere Ressourcen

Support

Bei Fragen oder Problemen:

  • GitHub Issues: Probleme melden 
  • Discussions: Community-Support auf GitHub Discussions
User TasksREST API