dataiku-utils 1.1.1

Helper utilities for Dataiku DSS project variables, date offsets and SQL-backed dataset columns.

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.

Method	Purpose
get_variable(key, scope="standard")	Read one variable.
exists_variable(key, scope="standard")	Check whether a key exists.
safe_update_scope_variables(values, scope="standard")	Merge values into one scope under lock.
safe_set_variables(variables)	Replace the full variable payload under lock.
safe_set_lock(lock_value)	Acquire the project lock.
release_lock(lock_value)	Release the lock when owned by the caller.

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.

Member	Purpose
key, scope, initial_value	Variable identity and default value.
value	Read or write the stored value. Creates the variable on first read when missing.
create_if_not_exists()	Ensure the variable exists without changing an existing value.
erase_value()	Set the value to None.

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

Member	Purpose
step	Offset applied by update().
update()	Read the current value, apply step, store the result.
apply_date_offset(date_value, offset)	Shift a date without touching project variables.
normalize_date(date_value)	Return a YYYY-MM-DD string.

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.

Method	Purpose
colnames_from_dataset(dataset)	Column names in schema order.
map_colname_coltype_from_dataset(dataset)	Ordered column-to-type mapping.
table_fullname_from_sql_dataset(dataset)	catalog.schema.table for SQL datasets.
create_dataset_with_read_partitions(src_dataset, partitions)	Dataset handle limited to selected partitions.

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.

Method	Purpose
find_min_value()	Lowest value found from start_min_value.
find_max_value()	Highest value found from start_max_value.
find_next_value(start_value)	Next value after start_value.
find_if_exists_value(value)	Return value when it exists in the column.

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:

Handler	Examples
python	python, pyspark, r, shell, spark_scala, sparkr
sql	sql_query, sql_script, hive, impala
visual	sync, shaker/prepare, sampling, split, join, grouping, topn
ml	training, prediction, scoring, evaluation, standalone_evaluation, time series and deep learning recipes
ai	prompt, embed_documents, nlp_llm_rag_embedding, LLM evaluation recipes
generic	Plugin recipes and unknown types, still exported through the raw API payload

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

Method	Purpose
to_json()	Export the full project as a dictionary.
to_json_string()	Same content as a JSON string.
dataset_to_json()	Export all datasets and their connections.
zone_to_json()	Export flow zones and object membership.
recipe_to_json()	Export all recipes with inputs, outputs and payload.
python_recipe_to_json()	Export Python recipes only.
sql_recipe_to_json()	Export SQL recipes only.
visual_recipe_to_json()	Export visual recipes (sync, filter, prepare, join, split, ...).
ml_recipe_to_json()	Export ML recipes (training, scoring, evaluation, ...).
ai_recipe_to_json()	Export LLM and RAG flow recipes.
scenario_to_json()	Export scenarios with triggers, steps and reporters.
streaming_endpoint_to_json()	Export streaming endpoints.
knowledge_bank_to_json()	Export retrievable knowledge banks.
flow_graph_to_json()	Export flow nodes and edges for cross-zone links.

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.

Method	Purpose
apply(document)	Restore a full project document.
apply_from_string(text)	Parse JSON and restore.
json_to_dataset(documents)	Restore datasets only.
json_to_recipe(documents)	Restore recipes with handler dispatch.
json_to_python_recipe(documents)	Restore Python recipes.
json_to_sql_recipe(documents)	Restore SQL recipes.
json_to_visual_recipe(documents)	Restore visual recipes.
json_to_ml_recipe(documents)	Restore ML recipes.
json_to_ai_recipe(documents)	Restore LLM and RAG recipes.
json_to_zone(documents)	Restore flow zones.
json_to_scenario(documents)	Restore scenarios.
json_to_streaming_endpoint(documents)	Restore streaming endpoints.
json_to_knowledge_bank(documents)	Restore knowledge banks.

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.