Govern Actions#

Actions are used to manually trigger user-defined scripts in Dataiku Govern. They are written in Python and exist at two levels: artifact and instance.

Here, we provide some use cases that demonstrate how to use actions.

Trigger a scenario on a design node from a govern artifact#

Scenarios defined at the project level in the Design node can be triggered directly from the Govern node using actions attached to project artifacts.

The logic implemented here is the following:

  • Retrieve the Dataiku projects associated with the Govern project.

  • Loop over the artifacts retrieving the associated project key and node ID and filtering out automation nodes.

  • Use DSSClient on each Dataiku artifact to retrieve the requested scenario and run it with the specified custom parameters.

Note

In the sample code below, the scenario is run synchronously, blocking the kernel until its completion. The scenario can also be run asynchronously using scenario.run(params)

from govern.core.artifact_action_handler import get_artifact_action_handler
import dataikuapi
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

handler = get_artifact_action_handler()

SCENARIO_ID = 'TESTPARAM'

DSS_NODES_CONFIG = {
    "Staging design": {
        "url": "https://design-node-url",
        "api_key": "SECRET_KEY"
    }
}

def get_dataiku_items_from_govern_item(govern_client, govern_item):
    definition = govern_item.get_definition()
    dku_items = definition.get_raw().get('fields', {}).get('dataiku_item', [])
    return [govern_client.get_artifact(artifact_id) for artifact_id in dku_items]

def trigger_scenario_on_node(node_id, project_key, scenario_id):
    if node_id not in DSS_NODES_CONFIG:
        logger.info(f"[Skipping] No credentials for Node ID: {node_id}")
        return

    creds = DSS_NODES_CONFIG[node_id]
    try:
        client = dataikuapi.DSSClient(creds["url"], creds["api_key"])
        project = client.get_project(project_key)
        scenario = project.get_scenario(scenario_id)
        params = handler.params

        logger.info(f"Triggering scenario '{SCENARIO_ID}' on project '{project_key}' and node '{node_id}'")

        # we can run the scenario asynchronously and not wait for the result
        # outcome = scenario.run(params)
        outcome = scenario.run_and_wait(params)
        logger.info(f"Scenario finished with outcome: {outcome.outcome}")

    except Exception as e:
        handler.status = "ERROR"
        handler.message = f"An error occurred: {e}"
        logger.error(f"An error occurred running the scenario: {e}")


def trigger_scenario_workflow():
    govern_client = handler.client

    govern_project = govern_client.get_artifact(handler.enrichedArtifact.artifact.id)

    linked_dku_artifacts = get_dataiku_items_from_govern_item(govern_client, govern_project)

    logger.info(f"Found {len(linked_dku_artifacts)} linked Dataiku projects.")

    for dku_artifact in linked_dku_artifacts:
        raw_fields = dku_artifact.get_definition().get_raw().get('fields', {})

        target_project_key = raw_fields.get('project_key')
        target_node_id = raw_fields.get('node_id')
        is_automation = raw_fields.get('automation_node')
        logger.info(f"Processing '{target_project_key}' on '{target_node_id}'")

        # Filter: Skip if it is an automation node
        if is_automation is True:
            logger.info(f"Skipping automation node project '{target_project_key}' and node '{target_node_id}'")
            continue

        # Process only Design nodes
        if target_project_key and target_node_id:
            trigger_scenario_on_node(target_node_id, target_project_key, SCENARIO_ID)



trigger_scenario_workflow()

Create a DSS Project with custom settings from a Govern Artifact#

Context#

This example demonstrates how to use Dataiku Govern for Project Pre-qualification before technical development begins.

  1. Blueprint & Pre-assessment: Users initiate the process by creating an Artifact from a Blueprint (e.g., “Project pre-qualification”). This stage allows stakeholders to document the feasibility, assess risks, and define necessary controls.

  2. Assertion of Readiness: When the assessment is complete, the user asserts readiness by enabling the checkbox field with id dss_project_creation. This acts as a validation gate.

  3. Trigger Creation Action: The user clicks the custom action button. If the validation gate is checked, the script automatically creates a project on the Dataiku Design node, carrying over the approved metadata.

Note

This script relies on three specific fields that should be present in the Blueprint Version:

  • description: a field of type TEXT

  • controls: a field of type TEXT list

  • dss_project_creation: a field of type BOOLEAN

Logic Implemented#

The logic implemented in the code sample is the following:

  • Retrieve the dss_project_creation, description and controls fields from the Govern artifact.

  • If the dss_project_creation field is True:

    • Connect to the remote Design node using dataikuapi.DSSClient.

    • Impersonate the artifact owner to ensure the project is created under their user profile.

      (Note: The user must exist on both the Govern and Design nodes for impersonation to succeed; otherwise, the script falls back to the Admin user.)

    • Create a new project using a key and name derived from the artifact.

    • Add specific tags and checklist to the project metadata.

    • Set the project status to “Draft”.

    • return the project link to be displayed in a success message

  • If the dss_project_creation field is False:

    • Set the handler status to ERROR and return a message prompting the user to confirm the creation.

Note

For more examples and API details, please refer to the Projects documentation.

from govern.core.artifact_action_handler import get_artifact_action_handler
import dataiku
import dataikuapi
import re
import logging

# import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

### CONFIGURATION TO CHANGE
DSS_URL = 'http://localhost:8086'
DSS_API_KEY = 'dkuaps-XXXXX'
### /CONFIGURATION


def create_checklist(author, items):
    checklist = {
        "title": "To-do list",
        "createdOn": 0,
        "items": []
    }
    for item in items:
        checklist["items"].append({
            "createdBy": author,
            "done": False,
            "stateChangedOn": 0,
            "text": item
        })
    return checklist

def impersonate_client(client, owner):
    try:
        # If it's an API key user, we usually cannot 'impersonate' them via get_user()
        # so we return the original client.
        if owner.startswith('api:'):
            logger.info(f"Request coming from api. keeping old client.")

            return client
        logger.info(f"Creating client to act as user {owner}.")
        return client.get_user(owner).get_client_as()
    except Exception as e:
        # Good practice: Log why it failed, fallback to original client
        logger.error(f"Impersonation failed for {owner}: {e}. Using original client.")
        return client


def create_custom_project(client,
                            owner,
                            project_key,
                            name,
                            custom_tags,
                            description,
                            checklist_items):

    acting_client = impersonate_client(client, owner)
    project = acting_client.create_project(project_key=project_key,
                                    name=name,
                                    owner=owner,
                                    description=description)

    logger.info(f"Project with key: {project.get_summary()['projectKey']} created.")

    # Add checklist
    metadata = project.get_metadata()
    metadata["checklists"]["checklists"].append(create_checklist(author=owner,
                                                                 items=checklist_items))
    # Add tags
    metadata["tags"] = custom_tags

    project.set_metadata(metadata)

    # Set default status to "Draft"
    settings = project.get_settings()
    settings.settings["projectStatus"] = "Draft"
    settings.save()
    logger.info(f"Project with key: {project.get_summary()['projectKey']} updated and set as draft.")

    return project

def clean_and_uppercase(input_string):
    # Remove all special characters, keeping only alphanumeric characters and spaces
    cleaned_string = re.sub(r'[^A-Za-z0-9]+', '', input_string)
    # Convert to uppercase
    return cleaned_string.upper()


handler = get_artifact_action_handler()
enrichedArtifact = handler.enrichedArtifact
artifact = enrichedArtifact.artifact

dss_project_tags = ["created from Govern"]
dss_project_controls = artifact.fields.get("controls")
if dss_project_controls:
    dss_project_tags.extend(dss_project_controls)

dss_project_checklist = [
    "Connect to data sources",
    "Clean, aggregate and join data",
    "Train ML model",
    "Evaluate ML model",
    "Deploy ML model to production"
    ]

dss_project_owner = handler.authCtxIdentifier
dss_project_name = artifact.name
dss_project_key = clean_and_uppercase(dss_project_name)
dss_project_description = artifact.fields.get("description")

checkbox_value = artifact.fields.get("dss_project_creation")
if (str(checkbox_value) == "True"):
    dss_client = dataikuapi.DSSClient(DSS_URL, DSS_API_KEY)

    project = create_custom_project(client=dss_client,
                                owner=dss_project_owner,
                                project_key=dss_project_key,
                                name=dss_project_name,
                                custom_tags=dss_project_tags,
                                description=dss_project_description,
                                checklist_items=dss_project_checklist)
    handler.status = "SUCCESS"
    handler.message = f"Project created with link: {DSS_URL}/projects/{project.get_summary()['projectKey']}"
else:
    handler.status = "ERROR"
    handler.message = "Please confirm the creation of the project by checking the box above."