# Workflows (internal API)

Base Workflow infrastructure.

See Workflows for a high level explanation of concepts used here.

class debusine.server.workflows.base.Workflow(work_request: WorkRequest)

  Bases: `BaseServerTask`, `Generic`

  Base class for workflow orchestrators.

  This is the base API for running `WorkflowInstance` logic.

  TASK_TYPE: TaskTypes = 'Workflow'

    The worker type must be suitable for the task type. TaskTypes.WORKER requires an external worker; TaskTypes.SERVER requires a Celery worker; TaskTypes.SIGNING requires a signing worker.

  callback(work_request: WorkRequest) → bool

    Perform an orchestration step.

    Called with the workflow callback work request (note that this is not the same as `self.work_request`) when the workflow node becomes ready to execute.

    Called with a `WorkRequest` of type internal/workflow to perform an orchestration step triggered by a workflow callback.

    This method is required to be idempotent: calling it multiple times with the same argument MUST result in the same `WorkRequest` structure as calling it once.

    Subclasses may provide `callback_{step.replace('-', '_')}` methods to implement workflow callbacks with the corresponding `step` set in their workflow data, or override `get_callback`.

    Returns:

      True if the callback succeeded, otherwise False.

  dynamic_task_data_type: type[DTD] = DTD

    Class used as the in-memory representation of dynamic task data.

  ensure_dynamic_data(task_database: TaskDatabaseInterface) → None

    Ensure that this workflow’s dynamic task data has been computed.

  classmethod from_name(name: str) → type[Workflow[Any, Any]]

    Instantiate a workflow by name.

  get_callback(step: str) → Callable[[], bool]

    Get the callback implementation for a given workflow step name.

    By default, subclasses may provide `callback_{step.replace('-', '_')}` methods to implement workflow callbacks with the corresponding `step` set in their workflow data. Alternatively, they may override this method to provide a more complex mapping from work request to callback implementation.

  get_input_artifacts() → list[InputArtifactSingle | InputArtifactMultiple]

    Return the list of input artifacts used by this workflow.

  get_input_artifacts_ids() → list[int]

    Return the list of artifact ids used by this workflow.

    Deprecated since version 2025-12: Use get_input_artifacts()

  lookup_singleton_collection(category: CollectionCategory, *, workspace: Workspace | None = None) → Collection

    Look up a singleton collection related to this workflow.

  name: ClassVar[str] = 'workflow'
  orchestrate_child(work_request: WorkRequest) → None

    Orchestrate a child work request in whatever way is appropriate.

    Silently skips non-pending/running work requests.

  abstractmethod populate() → None

    Create the initial WorkRequest structure.

    This is called once, when the workflow first becomes runnable. `validate_input()` will already have been called.

    This method is required to be idempotent: calling it multiple times with the same argument MUST result in the same `WorkRequest` structure as calling it once.

  static provides_artifact(work_request: WorkRequest, category: ArtifactCategory, name: str, *, data: dict[str, Any] | None = None, artifact_filters: dict[str, Any] | None = None) → None

    Indicate work_request will provide an artifact.

    Parameters:

      - work_request – work request that will provide the artifact

      - category – category of the artifact that will be provided

      - name – name of this item in the workflow’s internal collection

      - data – add it to the data dictionary for the event reaction

      - artifact_filters – for the update-collection-with-artifacts action, to allow workflows to add filtering

    Raises:

      LookupError – if a key in “data” starts with `promise_`

    Create an event reaction for `on_creation` adding a promise: this work request will create an artifact.

    Create an event reaction for `on_success` to update the collection with the relevant artifact.

  static requires_artifact(work_request: WorkRequest, lookup: int | str | LookupMultiple) → None

    Indicate that work_request requires input (lookup).

    Parameters:

      - work_request – for each lookup result call `work_request.add_dependency(promise["promise_work_request_id"])`

      - lookup – resolve the lookup and iterate over the results (for PROMISES only)

  task_data_type: type[TD] = WD

    Class used as the in-memory representation of task data.

  validate_input() → None

    Thorough validation of input data.

    This is run only once at workflow instantiation time, and can do slower things like database lookups to validate artifact or collection types.

  classmethod validate_runtime_parameters(data: dict[str, Any] | Literal[RuntimeParameter.ANY]) → None

    Validate WorkflowTemplate runtime_parameters.

  classmethod validate_static_parameters(data: dict[str, Any]) → None

    Validate WorkflowTemplate static_parameters.

  work_request_ensure_child_internal(*, task_name: str, workflow_data: WorkRequestWorkflowData, relative_priority: int = 0) → WorkRequest

    Create a child INTERNAL work request if one does not already exist.

    This assumes that `workflow_data.step` is unique among work request children with this task type and name, and that it is stable even if the workflow is orchestrated multiple times.

    Parameters:

      - task_name – the task name for the child work request.

      - task_data – the task data for the child work request.

      - workflow_data – the workflow data for the child work request.

      - relative_priority – if creating a new work request, set the base priority of the child to the effective priority of this workflow plus this relative priority.

    Returns:

      new or existing `WorkRequest`.

  work_request_ensure_child_server(*, task_name: str, task_data: BaseTaskData | None = None, workflow_data: WorkRequestWorkflowData) → WorkRequest

    Create a child SERVER work request if one does not already exist.

    This assumes that `workflow_data.step` is unique among work request children with this task type and name, and that it is stable even if the workflow is orchestrated multiple times.

    Parameters:

      - task_name – the task name for the child work request.

      - task_data – the task data for the child work request.

      - workflow_data – the workflow data for the child work request.

    Returns:

      new or existing `WorkRequest`.

  work_request_ensure_child_signing(*, task_name: str, task_data: BaseTaskData | None = None, workflow_data: WorkRequestWorkflowData) → WorkRequest

    Create a child SIGNING work request if one does not already exist.

    This assumes that `workflow_data.step` is unique among work request children with this task type and name, and that it is stable even if the workflow is orchestrated multiple times.

    Parameters:

      - task_name – the task name for the child work request.

      - task_data – the task data for the child work request.

      - workflow_data – the workflow data for the child work request.

      - relative_priority – if creating a new work request, set the base priority of the child to the effective priority of this workflow plus this relative priority.

    Returns:

      new or existing `WorkRequest`.

  work_request_ensure_child_wait(*, task_name: str, task_data: BaseTaskData | None = None, workflow_data: WorkRequestWorkflowData) → WorkRequest

    Create a child WAIT work request if one does not already exist.

    This assumes that `workflow_data.step` is unique among work request children with this task type and name, and that it is stable even if the workflow is orchestrated multiple times.

    Parameters:

      - task_name – the task name for the child work request.

      - task_data – the task data for the child work request.

      - workflow_data – the workflow data for the child work request.

    Returns:

      new or existing `WorkRequest`.

  work_request_ensure_child_worker(*, task_name: str, task_data: BaseTaskData | None = None, workflow_data: WorkRequestWorkflowData, relative_priority: int = 0) → WorkRequest

    Create a child WORKER work request if one does not already exist.

    This assumes that `workflow_data.step` is unique among work request children with this task type and name, and that it is stable even if the workflow is orchestrated multiple times.

    Parameters:

      - task_name – the task name for the child work request.

      - task_data – the task data for the child work request.

      - workflow_data – the workflow data for the child work request.

      - relative_priority – if creating a new work request, set the base priority of the child to the effective priority of this workflow plus this relative priority.

    Returns:

      new or existing `WorkRequest`.

  work_request_ensure_child_workflow(*, task_name: str, task_data: BaseTaskData | None = None, workflow_data: WorkRequestWorkflowData, relative_priority: int = 0) → WorkRequest

    Create a child WORKFLOW work request if one does not already exist.

    This assumes that `workflow_data.step` is unique among work request children with this task type and name, and that it is stable even if the workflow is orchestrated multiple times.

    Parameters:

      - task_name – the task name for the child work request.

      - task_data – the task data for the child work request.

      - workflow_data – the workflow data for the child work request.

      - relative_priority – if creating a new work request, set the base priority of the child to the effective priority of this workflow plus this relative priority.

    Returns:

      new or existing `WorkRequest`.

exception debusine.server.workflows.base.WorkflowRunError(work_request: WorkRequest, message: str, code: str)

  Bases: `Exception`

  Running a workflow orchestrator or callback failed.

  __init__(work_request: WorkRequest, message: str, code: str) → None

    Construct the exception.

exception debusine.server.workflows.base.WorkflowValidationError

  Bases: `Exception`

  Raised if a workflow fails to validate its inputs.

debusine.server.workflows.base.describe_exc(exception: Exception) → str

  User-facing description of exception.

debusine.server.workflows.base.orchestrate_workflow(work_request: WorkRequest) → bool

  Orchestrate a workflow in whatever way is appropriate.

  For a workflow callback, run `callback` and mark the work request as completed. For a workflow, run `populate` and unblock workflow children, but leave the workflow running until all its children have finished. For any other work request, log an error.

  Return False if any errors were logged, otherwise True.

Utility functions for workflows.

exception debusine.server.workflows.workflow_utils.ArtifactHasNoBinaryPackageName

  Bases: `Exception`

  Raised if it’s not possible to determine the artifact’s binary name.

debusine.server.workflows.workflow_utils.configure_for_overlay_suite(workflow: Workflow[Any, Any], *, extra_repositories: Sequence[ExtraDebusineRepository | ExtraExternalRepository] | None, vendor: str, codename: str, environment: int | str, backend: BackendType, architecture: str, try_variant: str) → list[ExtraDebusineRepository | ExtraExternalRepository] | None

  Return any needed extra repository to use an overlay suite.

debusine.server.workflows.workflow_utils.filter_artifact_lookup_by_arch(workflow: Workflow[Any, Any], lookup: LookupMultiple, architectures: set[str]) → LookupMultiple

  Filter an artifact lookup by architecture.

debusine.server.workflows.workflow_utils.follow_artifact_relations(artifact: Artifact, relation_type: Relations, category: ArtifactCategory) → QuerySet

  Follow relations from artifact to find artifacts of category.

debusine.server.workflows.workflow_utils.get_available_architectures(workflow: Workflow[Any, Any], *, vendor: str, codename: str) → set[str]

  Get architectures available for use with this vendor/codename.

debusine.server.workflows.workflow_utils.get_lookup_architectures(workflow: Workflow[Any, Any], lookup: LookupMultiple) → set[str]

  Return set with all the architectures in the artifacts from the lookup.

  The architectures are extracted from each lookup result using `lookup_result_architectures()`.

debusine.server.workflows.workflow_utils.get_source_package_names(results: Sequence[LookupResult], *, configuration_key: str, artifact_expected_categories: Collection[ArtifactCategory]) → list[str]

  Return a sorted list of source package names from results.

  It ensures that:

  - The `LookupResult` objects contain either an artifact or promise.

  - Artifacts belong to the artifact_expected_categories.

  - If `LookupResult` is a promise: extracts the name from the promise data `source_package_name`.

  Parameters:

    - results – A sequence of `LookupResult` objects representing artifacts to be processed. Each entry is expected to be either an artifact or a promise.

    - configuration_key – A string used by `ensure_artifact_categories()` for the exception message.

    - artifact_expected_categories – valid `ArtifactCategory` that artifacts must belong to.

  Returns:

    A sorted list of source package names.

debusine.server.workflows.workflow_utils.locate_debian_binary_packages(configuration_key: str, artifacts: Sequence[Artifact]) → list[Artifact]

  Find debian:binary-package artifacts for a workflow.

  Parameters:

    - configuration_key – The key in the workflow’s task data from which the given artifacts were looked up.

    - artifacts – A sequence of debian:binary-package or debian:upload artifacts.

  Returns:

    A list of artifacts: for each element of `artifacts`, if it is a debian:binary-package, return it; if it is a debian:upload, return all the related debian:binary-package artifacts.

debusine.server.workflows.workflow_utils.locate_debian_source_package(configuration_key: str, artifact: Artifact) → Artifact

  Find a debian:source-package artifact for a workflow.

  Parameters:

    - configuration_key – The key in the workflow’s task data from which the given artifact was looked up.

    - artifact – A debian:source-package or debian:upload artifact.

  Returns:

    If `artifact` is a debian:source-package, return it; if it is a debian:upload, return the related debian:source-package.

debusine.server.workflows.workflow_utils.locate_debian_source_package_lookup(workflow: Workflow[Any, Any], configuration_key: str, lookup: int | str) → int | str

  Return a lookup to a debian:source-package.

  If the specified lookup returns a debian:source-package, return it. If it returns a debian:upload, find the related debian:source-package and return a lookup to it.

debusine.server.workflows.workflow_utils.lookup_result_architectures(result: LookupResult) → set[str]

  Get binary package architectures present within LookupResult.

debusine.server.workflows.workflow_utils.lookup_result_artifact_category(result: LookupResult) → str

  Get artifact category from result of looking up an artifact.

  The result may be either an artifact or a promise.

debusine.server.workflows.workflow_utils.lookup_result_binary_package_name(result: LookupResult) → str

  Get binary package name from result of looking up an artifact.

debusine.server.workflows.workflow_utils.source_package(workflow: Workflow[Any, Any], *, configuration_key: str = 'source_artifact') → Artifact

  Retrieve the source package artifact from `configuration_key`.

  If the source artifact is a debian:upload, returns its debian:source-package.

debusine.server.workflows.workflow_utils.source_package_data(workflow: Workflow[Any, Any], *, configuration_key: str = 'source_artifact') → DebianSourcePackage

  Return source package artifact data for the workflow.