c8-client 8.7.2

Core library for Camunda 8 operations

c8-client Python Library

License

This project is licensed under the Apache License 2.0 with the Commons Clause restriction.

• ✅ You may use this library in commercial or private projects.
• ✅ You may modify, distribute, and embed it in your own products.
• ❌ You may not sell this library as a standalone product or offer it as a paid hosted service.
• ❌ You may not create a product whose primary value comes from this library's functionality and sell it.

See LICENSE for full details.

camunda_client.c8 Camunda 8 Module

Contains Camunda 8 (C8) API related features to handle jobs

camunda_client.c8.models

Contains C8 API related request and response data transfer object classes to structure data used for communication with the C8 API.

All models are based on a custom implementation camunda_client.utils.JsonBaseModel based on pydantic.BaseModel

camunda_client.c8.api

Contains functions to request Camunda 8 API endpoints.

Included API Endpoints:

Job API:

1. Activate Jobs camunda_client.c8.api.activate_jobs
2. Complete Job camunda_client.c8.api.complete_job
3. Mark Job as Failed camunda_client.c8.api.mark_job_as_failed
4. Throw Error for Job camunda_client.c8.api.throw_error_for_job

Process Instance API:

1. Create process instance camunda_client.c8.api.create_process_instance
2. Cancel process instance camunda_client.c8.api.cancel_process_instance
3. Migrate process instance camunda_client.c8.api.migrate_process_instance
4. Modify process instance camunda_client.c8.api.modify_process_instance

Message API:

1. Publish a message camunda_client.c8.api.publish_message
2. Correlate a message camunda_client.c8.api.correlate_message

Signal API:

1. Broadcast signal camunda_client.c8.api.broadcast_signal

Authentication API:

1. Generate a Token camunda_client.c8.auth.get_token

camunda_client.c8.worker

Camunda job worker can be defined by decorating a worker function with @camunda_worker and passing the job type:

# app.py
from camunda_client.c8.models import ActivatedJob
from camunda_client.c8.worker import camunda_worker


@camunda_worker(job_type="a-camunda-job-type")
def a_camunda_job_worker(job: ActivatedJob) -> dict:
    job_variables = job.variables
    # Worker processes...
    return {
        "output": "some output for the next task"
    }

To keep the application alive and to activate jobs for each registered @camunda_worker, a main application loop should be implemented:

# app.py
import time
import logging
import os

from dotenv import load_dotenv

import camunda_client.config as config
from camunda_client.c8.models import ActivatedJob
from camunda_client.c8.worker import (
    start_camunda_workers,
    stop_camunda_workers,
    camunda_worker,
    register_workers
)

log = logging.getLogger(__name__)


@camunda_worker(job_type="a-camunda-job-type")
def a_camunda_job_worker(job: ActivatedJob) -> dict:
    job_variables = job.variables
    # Worker processes...
    return {
        "output": "some output for the next task"
    }


def configure():
    load_dotenv()
    config.initialize_config(
        config=config.C8Config(
            auth=config.AuthConfig(
                selfManaged=config.SelfManagedAuthConfig(
                    clientId=os.getenv("CAMUNDA_CLIENT_ID"),
                    clientSecret=os.getenv("CAMUNDA_CLIENT_SECRET"),
                    oidcTokenUrl=os.getenv("CAMUNDA_TOKEN_URL")
                )
            ),
            api=config.ApiConfig(
                baseUrl=os.getenv("CAMUNDA_API_BASE_URL"),
                job=config.JobConfig(
                    worker="c8-example-worker",
                    timeout=5000,
                )
            )
        )
    )


def application_loop():
    """
    Primary application loop that starts the Camunda job workers and busy-waits
    until a Keyboard interruption (Ctrl+C) is received or an exception occurs.
    Then, it stops the workers and the server.
    """
    try:
        register_workers()
        start_camunda_workers()
        while True:
            time.sleep(2)
    except KeyboardInterrupt:
        log.info("Ctrl+C detected. Stopping application gracefully.")
    except Exception as e:
        log.critical(
            f"An unhandled exception occurred in the main application loop: {e}",
            exc_info=True,
        )
    finally:
        stop_camunda_workers()
        log.info("Application stopped.")

camunda_client.error Error Module

Contains camunda_client specific exceptions:

CamundaBusinessException: Can be thrown on business error in a @camunda_worker decorated function. Will cause a call to the Camunda 8 API 'Throw Error for Job' endpoint.

ServiceException: Will be thrown on technical errors (e.g. Camunda 8 API is not reachable). The activated job will be marked as failed.

SignatureException: Is thrown on initialization errors for @camunda_worker (e.g. decorated function does not accept ActivatedJob as parameter)

camunda_client.config Configuration Module

Handles the configuration for the camunda_client library. On application start, the configuration must be passed:

import camunda_client.config as config
import os

config.initialize_config(
    config=config.C8Config(
        auth=config.AuthConfig(
            selfManaged=config.SelfManagedAuthConfig(
                clientId=os.getenv("CAMUNDA_CLIENT_ID"),
                clientSecret=os.getenv("CAMUNDA_CLIENT_SECRET"),
                oidcTokenUrl=os.getenv("CAMUNDA_TOKEN_URL")
            )
        ),
        api=config.ApiConfig(
            baseUrl=os.getenv("CAMUNDA_API_BASE_URL"),
            job=config.JobConfig(
                worker="c8-example-worker",
                timeout=5000,
            )
        )
    )
)