dataiku-utils 1.0.0

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)

---

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.