Govern Hooks

Hooks are used to automate actions related to artifacts in Dataiku Govern. They are written in Python and will be run during the artifacts lifecycle phases: CREATE, UPDATE, or DELETE. Which phases are selected to run a hook is configurable.

This page gathers multiple hooks samples for different use cases.

Define fields to be mandatory for the completion of a workflow step

In Govern, fields are not attached to workflow steps by default. Setting a field as required makes it required at all time in any state. In some use cases, it may be necessary to check the requirement of some fields only when completing a step (setting a step as finished). In this case, it’s possible to add a hook that defines the fields that are mandatory for each workflow step and checks if those fields are set when setting the corresponding workflow step as finished.

Note that with this hook, 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"]
    }
)