Automate actions with Govern Hooks

Hooks are used to automate actions related to artifacts in Dataiku Govern. They are written in Python and run during specified artifact lifecycle phases, including:

• CREATE

• UPDATE

• DELETE

Here, we provide a use case that demonstrates how to use hooks.

Make a field mandatory for a workflow step

By default, no fields in Dataiku Govern are mandatory for completing a workflow step. However, there might be a case where you want to check if a field is populated before a workflow step is marked as finished.

In this situation, hooks can be used to define and automatically check this condition.

Note

In the sample hook below, the fields that are mandatory only for setting a workflow step as finished should not be set as “Required” in the field configuration.

from govern.core.handler import get_handler

def check_mandatory_step_fields(hookHandler, step_mandatory_field_ids):
    def get_step_index(step_definitions, step_id):
        for i, v in enumerate(step_definitions):
            if v.get('id', None) == step_id:
                return i

    def field_ids_from_view_component(view_component):
        if view_component is None:
            return []
        field_id = view_component.get('fieldId', '')
        if len(field_id) > 0:
            return [field_id]
        if view_component.get('type', '') == 'container':
            layout = view_component.get('layout', None)
            if layout is not None:
                if layout.get('type', '') == 'sequential':
                    ret = []
                    for vc in layout.get('viewComponents', []):
                        ret = ret + field_ids_from_view_component(vc)
                    return ret
        return []

    def field_ids_from_step_id(nea, step_id):
        uiStepDefinition = nea.blueprintVersion.json.get('uiDefinition', {}).get('uiStepDefinitions', {}).get(step_id, None)
        if uiStepDefinition is None:
            return []
        view_id = uiStepDefinition.get('viewId', '')
        if len(view_id) <= 0:
            return []
        view = nea.blueprintVersion.json.get('uiDefinition', {}).get('views', {}).get(view_id, {})
        if view is None or view.get('type', '') != 'card':
            return []
        return field_ids_from_view_component(view.get('viewComponent', None))

    # 2/ Then it retrieves all the associated fields by looking at the configuration of the view associated with the workflow step:
    def check_step(nea, step_id, mandatory_fields):
        field_ids = field_ids_from_step_id(nea, step_id)
        # 3/ Finally, looping through the fields attached to the workflow step, it checks the ones defined as mandatory for this step and raises an error if those fields are not set:
        for field_id in field_ids:
            if field_id in mandatory_fields:
                field_value = nea.artifact.fields.get(field_id, None)
                if field_value is None or (isinstance(field_value, str) and len(field_value) == 0) or (isinstance(field_value, list) and len(field_value) == 0):
                    handler.fieldMessages[field_id] = "field is mandatory in step id: " + step_id
                    handler.status = "ERROR"

    # 1/ This hook first aims to detect that the user is trying to set a specific workflow step as finished and retrieve the corresponding step id:
    if hookHandler.hookPhase == 'DELETE':
        return
    nea = hookHandler.newEnrichedArtifact
    if nea is None:
        return

    stepDefinitions = nea.blueprintVersion.json.get('workflowDefinition', {}).get('stepDefinitions', [])
    if len(stepDefinitions) <= 0:
        return

    step_id = nea.artifact.json.get('status', {}).get('stepId', '')
    step_index = get_step_index(stepDefinitions, step_id)

    for i in range(0, step_index):
        previous_step_id = stepDefinitions[i].get('id', '')
        if not previous_step_id in step_mandatory_field_ids:
            continue
        check_step(nea, previous_step_id, step_mandatory_field_ids[previous_step_id])

handler = get_handler()

# 4/ The way to attach fields to a workflow step is as follows:
check_mandatory_step_fields(
    handler,
    {
        # Each step and corresponding fields are identified by their ids.
        # You can add as many mandatory fields for any workflow step as you wish.
        "exploration": ["mandatory_exploration"],
        "qualification": ["mandatory_qualification", "mandatory_qualification_2"],
        "progress": ["mandatory_ref_progress"]
    }
)