dataiku-utils

dataiku-utils is a small Python library for Dataiku DSS projects. It wraps project variables with a locking layer, exposes a simple object API around reads and writes, and includes helpers for date offsets and SQL-backed column lookups.

The code is meant to run inside Dataiku recipes, scenarios or project libraries where the dataiku Python module is available.

Installation

pip install dataiku-utils

Local install from a source archive:

pip install dataiku_utils-1.0.0.tar.gz

Runtime requirement

The dataiku module is provided by Dataiku DSS at runtime. It is not installed from PyPI and is therefore not declared as a package dependency.

Why this package exists

Dataiku project variables are shared at project level. When two scenario steps run in parallel and update the same variable, one write can overwrite the other. The classes below reduce that risk with a lightweight lock and keep common update patterns in one place.

LogBase

Base class used by every public helper. It provides one logger per class.

import logging
from dataiku_utils import ProjectVariables

ProjectVariables.logger().setLevel(logging.DEBUG)

ProjectVariables

Low-level access to project variables with locking helpers.

from dataiku_utils import ProjectVariables

current = ProjectVariables.get_variable("last_processed_date")

ProjectVariables.safe_update_scope_variables(
    {"last_processed_date": "2026-01-31"},
    scope="standard",
)

The lock is stored in the local scope under the key lock. Waits time out after max_retries attempts (default 40).

SimpleValueVariablesUtils

Wraps a single project variable behind a value attribute.

from dataiku_utils import SimpleValueVariablesUtils

variable = SimpleValueVariablesUtils(
    key="processing_date",
    initial_value="2026-01-01",
)

variable.create_if_not_exists()
print(variable.value)
variable.value = "2026-01-02"

Writes compare against the live value in Dataiku before updating, which avoids skipping an update when a parallel step changed the variable in the meantime.

ValueVariablesUtils

Extends SimpleValueVariablesUtils with an update() method. Subclasses implement next_value() to compute the value to store.

from dataiku_utils import ValueVariablesUtils

class CounterVariable(ValueVariablesUtils):
    def next_value(self):
        return int(self.value) + 1

counter = CounterVariable(key="run_count", initial_value=0)
counter.create_if_not_exists()
counter.update()

update() reloads the current value from Dataiku before calling next_value().

WatchedValueVariablesUtils

Adds a companion boolean variable that records whether update() changed the main value. The status key defaults to {key}__is_updated.

from dataiku_utils import WatchedValueVariablesUtils

class MyWatchedVariable(WatchedValueVariablesUtils):
    def next_value(self):
        return "next"

variable = MyWatchedVariable(key="status_marker")
variable.create_if_not_exists()

if variable.update():
    print("Value changed")
print(variable.is_updated_variable.value)

DateValueVariableUtils

Date variable with a fixed step. The step format is {+|-}{number}{D|M|W}:

• D — days
• M — months
• W — weeks

from dataiku_utils import DateValueVariableUtils

variable = DateValueVariableUtils(
    key="calcul_dt",
    initial_value="2018-01-01",
    step="+1D",
)

variable.create_if_not_exists()
variable.update()          # 2018-01-01 -> 2018-01-02
variable.value = "2020-06-15"

DateMultipleValueVariableUtils

Keeps a base date variable and a set of derived variables in sync. Each entry in refs maps a dependent variable name to an offset relative to the base value.

from dataiku_utils import DateMultipleValueVariableUtils

variable = DateMultipleValueVariableUtils(
    key="calcul_dt",
    initial_value="2018-01-01",
    step="+1D",
    refs={
        "calcul_dt_minus_6M": "-6M",
        "calcul_dt_plus_3W": "+3W",
    },
)

variable.create_if_not_exists()
variable.update()
variable.value = "2019-05-10"

After create_if_not_exists(), update() or a direct assignment to value, all dependent variables are recalculated from the base date.

DatasetUtils

Read metadata from Dataiku dataset objects.

import dataiku
from dataiku_utils import DatasetUtils

dataset = dataiku.Dataset("input_dataset", ignore_flow=True)

columns = DatasetUtils.colnames_from_dataset(dataset)
table_name = DatasetUtils.table_fullname_from_sql_dataset(dataset)

TableColumnValueUtils

Discovers values in a SQL-backed column through small LIMIT 1 queries. This approach is often faster than filtering the full table on large datasets, especially when the column is not the partition key.

from dataiku_utils import TableColumnValueUtils

lookup = TableColumnValueUtils(
    dataset_name="input_dataset",
    colname="event_date",
    coltype="DATE",
    start_min_value="2026-01-01",
    start_max_value="2026-12-31",
    queries_common_condition="country = 'FR'",
)

minimum = lookup.find_min_value()
next_date = lookup.find_next_value("2026-01-15")

TableColumnValueVariablesUtils

Combines TableColumnValueUtils and ValueVariablesUtils. The project variable starts at the column minimum and update() moves to the next available value.

from dataiku_utils import TableColumnValueVariablesUtils

variable = TableColumnValueVariablesUtils(
    value_key="last_processed_value",
    dataset_name="input_dataset",
    colname="event_date",
    coltype="DATE",
    start_min_value="2026-01-01",
    start_max_value="2026-12-31",
)

variable.create_if_not_exists()
updated = variable.update()

TableColumnWatchedValueVariablesUtils

Same as TableColumnValueVariablesUtils, plus the __is_updated companion variable from WatchedValueVariablesUtils.

from dataiku_utils import TableColumnWatchedValueVariablesUtils

variable = TableColumnWatchedValueVariablesUtils(
    value_key="last_processed_value",
    dataset_name="input_dataset",
    colname="event_date",
    coltype="DATE",
    start_min_value="2026-01-01",
    start_max_value="2026-12-31",
)

variable.create_if_not_exists()
if variable.update():
    print("Advanced to", variable.value)

Project JSON import and export

Convert a full Dataiku DSS v13 project to JSON and restore it later. The export stores the raw API payloads used by dataikuapi, which keeps recipe scripts, visual settings and ML configuration intact across environments.

Covered flow objects:

• Zones, datasets, recipes, scenarios
• Saved models, model evaluation stores, managed folders
• Streaming endpoints and knowledge banks used by RAG recipes

Notebook and code studio content is not exported.

Install the optional API client:

pip install dataiku-utils[project-json]

Recipe families handled by the built-in registry:

Each recipe document contains definition_and_payload, which is the exact body returned by the DSS recipe API. This is the field used on import.

DataikuProjectToJson

from dataiku_utils import DataikuProjectToJson

exporter = DataikuProjectToJson(
    project_key="MY_PROJECT",
    host="https://dss.example.com",
    api_key="your-api-key",
)

document = exporter.to_json()
text = exporter.to_json_string()

python_recipes = exporter.python_recipe_to_json()
scenarios = exporter.scenario_to_json()

Each recipe document contains a handler field (python, sql, visual, ml, ai or generic) so the importer knows which restore class to call.

JsonToDataikuProject

Restores objects in dependency order: datasets and ML artifacts, knowledge banks, recipes, zones, then scenarios.

from dataiku_utils import JsonToDataikuProject

importer = JsonToDataikuProject(
    project_key="TARGET_PROJECT",
    host="https://dss.example.com",
    api_key="your-api-key",
    overwrite=True,
)

importer.apply(document)

Specialized converter classes are also available directly from dataiku_utils.project_json, for example DatasetToJson, JsonToDataset, RecipeToJson, ScenarioToJson, ZoneToJson and their JsonTo* counterparts.

API documentation

Source files use NumPy-style docstrings. Build the HTML reference with Sphinx:

chmod +x docs/build.sh
./docs/build.sh

Open docs/build/html/index.html in a browser.

Development

Build the package:

python -m pip install --upgrade build twine
python -m build
twine check dist/*

Publish to PyPI:

chmod +x pypi.sh
./pypi.sh

TestPyPI:

PYPI_REPOSITORY=testpypi ./pypi.sh

Push to git:

chmod +x gitpush.sh
./gitpush.sh

License

Proprietary.