Skip to Content
Client LibrariesPythonPython Client

Python Client

Der Python Client ist eine vollständige Python-Library für die Integration mit der 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 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 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 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:

User TasksREST API