Migration

This module provides access to the migration REST api of Camunda.

MigrationInstruction

class pycamunda.migration.MigrationInstruction(source_activity_ids: Tuple[str], target_activity_ids: Tuple[str], update_event_trigger: bool)

Data class of migration instruction as returned by the REST api of Camunda.

MigrationPlan

class pycamunda.migration.MigrationPlan(source_process_definition_id: str, target_process_definition_id: str, instructions: Tuple[pycamunda.migration.MigrationInstruction])

Data class of migration plan as returned by the REST api of Camunda.

InstructionReport

class pycamunda.migration.InstructionReport(instruction: pycamunda.migration.MigrationInstruction, failures: Tuple[str])

Data class of instruction report as returned by the Camunda REST api.

Generate

class pycamunda.migration.Generate(url: str, source_process_definition_id: str, target_process_definition_id: str, update_event_triggers: bool = False)

Generates a migration plan for 2 process definitions. The generated plan contains instructions that map equal activities between the process definitions.

Parameters:
  • url – Camunda Rest engine URL.
  • source_process_definition_id – Id of the source process definition for the migration.
  • target_process_definition_id – Id of the target process definition for the migration.
  • update_event_triggers – Whether instructions between events should be configured to update the event triggers.
__call__(*args, **kwargs) → pycamunda.migration.MigrationPlan

Send the request.

Validate

class pycamunda.migration.Validate(url: str, source_process_definition_id: str, target_process_definition_id: str)

Validates a migration plan without executing it.

Parameters:
  • url – Camunda Rest engine URL.
  • source_process_definition_id – Id of the source process definition for the migration.
  • target_process_definition_id – Id of the target process definition for the migration.
__call__(*args, **kwargs) → Tuple[pycamunda.migration.InstructionReport]

Send the request.

add_instruction(source_activity_ids: Iterable[str], target_activity_ids: Iterable[str], update_event_trigger: bool = False)

Add an instruction to validate.

Parameters:
  • source_activity_ids – The mapped activity ids from the source process definition.
  • target_activity_ids – The mapped activity ids from the target process definition.
  • update_event_trigger – Whether event triggers will be updated during migration.
classmethod from_migration_plan(url: str, migration_plan: pycamunda.migration.MigrationPlan) → pycamunda.migration.Validate

Create an instance of Validate using a MigrationPlan instance.

Parameters:
  • url – Camunda REST engine url.
  • migration_plan – The migration plan to create the Validate instance from.
Returns:

Validate instance.

Execute

class pycamunda.migration.Execute(url: str, source_process_definition_id: str, target_process_definition_id: str, process_instance_ids: Iterable[str], skip_custom_listeners: bool = False, skip_io_mappings: bool = False, async_: bool = False)

Executes a migration plan.

Parameters:
  • url – Camunda Rest engine URL.
  • source_process_definition_id – Id of the source process definition for the migration.
  • target_process_definition_id – Id of the target process definition for the migration.
  • process_instance_ids – Process instance ids to migrate.
  • skip_custom_listeners – Whether execution listeners should be invoked.
  • skip_io_mappings – Whether input / output mappings should be executed.
  • async – Whether to run this request asynchronously.
__call__(*args, **kwargs) → Optional[pycamunda.batch.Batch]

Send the request.

add_instruction(source_activity_ids: Iterable[str], target_activity_ids: Iterable[str], update_event_trigger: bool = False)

Add an instruction to execute.

Parameters:
  • source_activity_ids – The mapped activity ids from the source process definition.
  • target_activity_ids – The mapped activity ids from the target process definition.
  • update_event_trigger – Whether event triggers will be updated during migration.
classmethod from_migration_plan(url: str, migration_plan: pycamunda.migration.MigrationPlan, process_instance_ids: Iterable[str], skip_custom_listeners: bool = False, skip_io_mappings: bool = False, async_: bool = False) → pycamunda.migration.Execute

Create an instance of Execute using a MigrationPlan instance.

Parameters:
  • url – Camunda REST engine url.
  • migration_plan – The migration plan to create the Validate instance from.
  • process_instance_ids – Process instance ids to migrate.
  • skip_custom_listeners – Whether execution listeners should be invoked.
  • skip_io_mappings – Whether input / output mappings should be executed.
  • async – Whether to run this request asynchronously.
Returns:

Execute instance.