1 of 11

Experimental Functions

SDK experimental functions are in early development and are not fully tested. These functions are light-weight wrappers of the API endpoints, and are mostly interacting via JSONs/dicts.

Function names and signature are expected to be changed when they are elevated to standard SDK functions in the future.

create_test

Create a new Test Spec

dbnl.experimental.create_test(
    *,
    test_spec_dict: Dict[str, Any]
) -> Dict[str, Any]:

Parameters

Arguments

Description

test_spec_dict

A dictionary of the expected Test Spec schema

Returns

Type

Description

Dict[str, Any]

The JSON dict of the created Test Spec object. The return JSON will contain the id of the Test Spec.

Test Spec JSON Schema

{
  "project_id": string,

  // Test data
  "name": string (project-unique),
  "description": string?,
  "statistic_name": string,
  "statistic_params": map[string, any],
  "statistic_inputs": list[
    {
      "select_query_template": {
        "select": string // a column or a function on column(s)
      }
    }
  ],
  "assertion": {
    "name": string,
    "params": map[string, any]
  },
  "tag_ids": string[]?
}

get_tests

Retrieve all Tests executed in the given Test Session.

dbnl.experimental.get_tests(
    *,
    test_session_id: str,
) -> List[Dict[str, Any]]:

Parameters

Arguments

Description

test_session_id

The Test Session ID associated with all the tests.

Returns

Type

Description

List[Dict[str, Any]]

A list of Test JSONs

Test JSON Schema

{
  // Test metadata
  "id": string,
  "org_id": string,
  "created_at": timestamp,
  "updated_at": timestamp,

  "test_session_id": string,

  // Test data
  "author_id": string,
  "value": any?,
  "failure": string?,
  "status": enum(PENDING, RUNNING, PASSED, FAILED),
  "started_at": timestamp?,
  "completed_at": timestamp?,

  // Test Spec data
  "test_spec_id": id,
  "name": string,
  "description": string?,
  "statistic_name": string,
  "statistic_params": map[string, any],
  "assertion": {
    "name": string,
    "params": map[string, any]
    "status": enum(...),
    "failure": string?
  },
  "statistic_inputs": list[
    {
      "select_query_template": {
        "select": string 
      }
    }
  ],
  "tag_ids": string[]?,
}

get_test_sessions

Retrieve all Test Sessions for a given Project

dbnl.experimental.get_test_sessions(
    *,
    project: Project,
) -> list[]:

Parameters

Arguments

Description

project

The dbnl Project associated with all the TestSessions.

Returns

Type

Description

List[TestSession]

A list of in the Project.

wait_for_test_session

Retrieve the completed TestSession

dbnl.experimental.wait_for_test_session(
    *,
    test_session: TestSession
    timeout_s: int = 180,
) -> :

Parameters

Arguments

Description

test_session

The dbnl to wait for completion.

timeout_s

An optional timeout (in seconds) parameter for waiting for the TestSession to complete. An exception will be raised if the TestSession did not complete before timeout_s has elapsed.

Returns

Type

Description

TestSession

The completed TestSession

Example

get_or_create_tag

Retrieve the specified dbnl Test Tag or create a new one if it does not exist

dbnl.experimental.get_or_create_tag(
    *,
    project_id: str,
    name: str,
    description: Optional[str] = None,
) -> Dict[str, Any]:

Parameters

Arguments

Description

project_id

The dbnl Project that this Test Tag is associated with.

name

The name of the Test Tag to be retrieved. If a Tag with the name does not exist, it will create a new Tag. Tag names must be unique.

description

An optional description for the Tag. Descriptions are limited to 255 characters.

Returns

Type

Description

Dict[str, Any]

Test Tag JSON

Test Tag JSON Schema

{
  // Tag metadata
  "id": string,
  "org_id": string,
  "created_at": timestamp,
  "updated_at": timestamp,
  
  // Tag data
  "name": string,
  "author_id": string,
  "description": string?,
  "project_id": string,
}

prepare_incomplete_test_spec_payload

Formats an incomplete Test Spec JSON

dbnl.experimental.prepare_incomplete_test_spec_payload(
    *,
    test_spec_dict: IncompleteTestSpecDict,
    project_id: Optional[str] = None,
) -> Dict[str, Any]:

Parameters

Arguments

Description

test_spec_dict

A dictionary of the . Fields that might be missing in this dict include project_id and tag_ids

project_id

The ID of the dbnl Project this Test Spec will be created for.

Returns

Type

Description

TestSpecDict

An acceptable dict Test Spec schema.

create_test_recalibration_session

def create_test_recalibration_session(
  *,
  test_session: TestSession,
  feedback: str,
  test_ids: Optional[list[str]] = None,
) ->

If a TestSession has completed and some of the generated Tests had incorrect outcomes, you can tell Distributional that these tests should have had different results.

Only applies to Tests generated by Distributional, not custom, user-defined Tests.

Parameters

Arguments

Description

Returns

Type

Description

Limitations

If some generated Tests failed when they should have passed, and some passed when they should have failed, the user will need to submit 2 separate calls, for each feedback result.

wait_for_test_recalibration_session

def wait_for_test_recalibration_session(
    *,
    test_recalibration_session: TestRecalibrationSession,
    timeout_s: int = 180,
) -> TestRecalibrationSession

Parameters

Arguments

Description

test_recalibration_session

The dbnl to wait for completion.

timeout_s

An optional timeout (in seconds) parameter for waiting for the TestRecalibrationSession to complete. An exception will be raised if the TestRecalibrationSession did not complete before timeout_s has elapsed.

Returns

Type

Description

TestRecalibrationSession

The completed TestRecalibrationSession

create_test_generation_session

def create_test_generation_session(
    *,
    run: Run,
    columns: Optional[list[str | dict[Literal["name"], str]]] = None,
) -> TestGenerationSession

If a TestSession has completed and some of the generated Tests had incorrect outcomes, you can tell Distributional that these tests should have had different results.

Only applies to Tests generated by Distributional, not custom Tests

Parameters

Arguments

Description

run

columns

Either a list of column names or a list of dictionaries with only 1 key, "name". These names must refer to the columns in the Run that dbnl will use to generate Tests. If None, then all columns in the Run will be used.

Returns

Type

Description

TestGenerationSession

The TestGenerationSession object that was created. The status can be checked to determine when test generation is complete.

Examples

The following 2 examples are equivalent:

run = dbnl.get_run(...)
dbnl.experimental.create_test_generation_session(
  run=run,
  columns=["col1", "col4"],
)

run = dbnl.get_run(...)
dbnl.experimental.create_test_generation_session(
  run=run,
  columns=[
    {"name": "col1"},
    {"name": "col4"},
  ],
)

To generate tests for all columns, leave columns out:

run = dbnl.run(...)
dbnl.experimental.create_test_generation_session(
  run=run,
)

wait_for_test_generation_session

Wait for a Test Generation Session to finish

def wait_for_test_generation_session(
    *,
    test_generation_session: TestGenerationSession,
    timeout_s: int = 180,
) -> TestGenerationSession

Parameters

Arguments

Description

test_generation_session

The dbnl to wait for completion.

timeout_s

An optional timeout (in seconds) parameter for waiting for the TestGenerationSession to complete. An exception will be raised if the TestGenerationSession did not complete before timeout_s has elapsed.

Returns

Type

Description

TestGenerationSession

The completed TestGenerationSession