Workflows

# From workflow.py:69-86
success_nodes = models.ManyToManyField('self', blank=True, symmetrical=False, related_name='%(class)ss_success')
failure_nodes = models.ManyToManyField('self', blank=True, symmetrical=False, related_name='%(class)ss_failure')
always_nodes = models.ManyToManyField('self', blank=True, symmetrical=False, related_name='%(class)ss_always')
all_parents_must_converge = models.BooleanField(
    default=False, 
    help_text=_("If enabled then the node will only run if all of the parent nodes have met the criteria to reach this node")
)

# From workflow.py:183-188
identifier = models.CharField(
    max_length=512,
    default=uuid4,
    blank=False,
    help_text=_('An identifier for this node that is unique within its workflow. It is copied to workflow job nodes corresponding to this node.'),
)

# From workflow.py:297-388
def get_job_kwargs(self):
    """
    In advance of creating a new unified job as part of a workflow,
    this method builds the attributes to use
    """
    data = {}
    wj_special_vars = {}
    wj_special_passwords = {}
    ujt_obj = self.unified_job_template
    if ujt_obj is not None:
        node_prompts_data = self.prompts_dict(for_cls=ujt_obj.__class__)
        wj_prompts_data = self.workflow_job.prompts_dict(for_cls=ujt_obj.__class__)
        # ...
        accepted_fields, ignored_fields, errors = ujt_obj._accept_or_ignore_job_kwargs(**node_prompts_data)
        data.update(accepted_fields)

Workflow extra_vars > Node extra_vars > Job Template extra_vars

- name: Register artifact
  ansible.builtin.set_stats:
    data:
      deployment_id: "{{ deployment.id }}"
      deployment_url: "https://app.example.com/{{ deployment.id }}"

# From workflow.py:345-355
aa_dict = {}
is_root_node = True
for parent_node in self.get_parent_nodes():
    is_root_node = False
    aa_dict.update(parent_node.ancestor_artifacts)
    if parent_node.job:
        aa_dict.update(parent_node.job.get_effective_artifacts(parents_set=set([self.workflow_job_id])))
if aa_dict and not is_root_node:
    self.ancestor_artifacts = aa_dict
    self.save(update_fields=['ancestor_artifacts'])

# From workflow.py:236-241
def create_approval_template(self, **kwargs):
    approval_template = WorkflowApprovalTemplate(**kwargs)
    approval_template.save()
    self.unified_job_template = approval_template
    self.save(update_fields=['unified_job_template'])
    return approval_template

# From workflow.py:901-917
def approve(self, request=None):
    self.status = 'successful'
    self.approved_or_denied_by = get_current_user()
    self.save()
    self.send_approval_notification('approved')
    self.websocket_emit_status(self.status)
    ScheduleWorkflowManager().schedule()
    return reverse('api:workflow_approval_approve', kwargs={'pk': self.pk}, request=request)

def deny(self, request=None):
    self.status = 'failed'
    self.approved_or_denied_by = get_current_user()
    self.save()
    self.send_approval_notification('denied')
    self.websocket_emit_status(self.status)
    ScheduleWorkflowManager().schedule()
    return reverse('api:workflow_approval_deny', kwargs={'pk': self.pk}, request=request)

# From workflow.py:64-70
unified_job_template = models.ForeignKey(
    'UnifiedJobTemplate',
    related_name='%(class)ss',
    blank=True,
    null=True,
    default=None,
    on_delete=models.SET_NULL,
)

# From workflow.py:382-388
if 'job_slice' in self.ancestor_artifacts and is_root_node:
    data['_eager_fields']['allow_simultaneous'] = True
    data['_eager_fields']['job_slice_number'] = self.ancestor_artifacts['job_slice']
    data['_eager_fields']['job_slice_count'] = self.workflow_job.workflow_job_nodes.count()
    data['_prevent_slicing'] = True

GET /api/v2/workflow_job_templates/

POST /api/v2/workflow_job_templates/
Content-Type: application/json

{
  "name": "Deploy Application",
  "description": "Full deployment workflow",
  "organization": 1,
  "extra_vars": "---\napp_version: 1.0.0",
  "ask_variables_on_launch": true
}

POST /api/v2/workflow_job_templates/{id}/workflow_nodes/
Content-Type: application/json

{
  "unified_job_template": 5,
  "identifier": "deploy-backend",
  "extra_data": {
    "service": "backend"
  }
}

POST /api/v2/workflow_job_template_nodes/{id}/success_nodes/
Content-Type: application/json

{
  "id": 42
}

POST /api/v2/workflow_job_template_nodes/{id}/create_approval_template/
Content-Type: application/json

{
  "name": "Deploy to Production?",
  "description": "Approve deployment to production environment",
  "timeout": 3600
}

POST /api/v2/workflow_job_templates/{id}/launch/
Content-Type: application/json

{
  "extra_vars": {
    "app_version": "1.2.3",
    "environment": "production"
  }
}

POST /api/v2/workflow_approvals/{id}/approve/

POST /api/v2/workflow_approvals/{id}/deny/

# From workflow.py:536-561
@property
def notification_templates(self):
    base_notification_templates = NotificationTemplate.objects.all()
    error_notification_templates = list(base_notification_templates.filter(unifiedjobtemplate_notification_templates_for_errors__in=[self]))
    started_notification_templates = list(base_notification_templates.filter(unifiedjobtemplate_notification_templates_for_started__in=[self]))
    success_notification_templates = list(base_notification_templates.filter(unifiedjobtemplate_notification_templates_for_success__in=[self]))
    approval_notification_templates = list(base_notification_templates.filter(workflowjobtemplate_notification_templates_for_approvals__in=[self]))