client.graph

An interview flow is the branching alternative to the linear question list. See the Interview flows guide for concepts and a worked example. Flows and patch operations can be plain dicts in the API’s JSON shape, or the typed FlowDocument, FlowNode, FlowEdge and PatchOp models. On the models, from and class are spelled from_ and class_.

get()

client.graph.get(study_id) -> GraphOut

The study’s flow, its graph_version, whether it’s the active script (is_active), and whether it last compiled cleanly (has_compiled_ir). flow is None if the study has no flow yet. Returns GraphOut.

set()

client.graph.set(study_id, *, flow, expected_graph_version: int | None = None) -> GraphCheckOut

Replace the whole flow. Nothing is saved unless the flow is valid: on any problem, the call raises ValidationError (with every problem in e.body) and the stored flow is untouched.

FlowDocument | dict

required

The complete flow.

int

The graph_version you last read. If the flow has been saved since, the write is refused with ConflictError. Omit to overwrite unconditionally.

Returns GraphCheckOut.

patch()

client.graph.patch(study_id, *, ops, expected_graph_version: int | None = None) -> GraphCheckOut

Apply targeted edits to the stored flow. Operations run in order, and the end state is validated as a whole: nothing is saved unless it’s valid.

Sequence[PatchOp | dict]

required

The edits, in order.

{"op": "add_node", "node": {...}}
{"op": "update_node", "id": "q_pets", "config": {"text": "New wording?"}}
{"op": "remove_node", "id": "q_pets"}
{"op": "add_edge", "edge": {"from": "q1", "to": "end"}}
{"op": "update_edge", "from": "branch", "to": "q2", "priority": 1}
{"op": "remove_edge", "from": "q1", "to": "end"}

update_node merges config: keys you omit keep their value, and None clears one.

int

Optimistic concurrency, as for set().

Returns GraphCheckOut.

check()

client.graph.check(study_id, *, flow=None) -> GraphCheckOut

Validate a flow and report every problem, without changing anything. Omit flow to re-check the stored one.

report = client.graph.check(study.id, flow=draft_flow)
for error in report.errors or []:
    print(error.node_id, error.code, error.message)   # e.g. "branch no_default ..."

Returns GraphCheckOut. When valid, compiled holds the derived variables table and execution order.

import_questions()

client.graph.import_questions(study_id) -> GraphCheckOut

Build and save a straight-line flow from the study’s linear questions. This replaces any existing flow. Each question captures its answer as q1, q2, and so on. It doesn’t change the interview mode; call activate() when ready. Returns GraphCheckOut.

activate()

client.graph.activate(study_id) -> GraphModeOut

Switch the study to graph mode, so interviews follow the stored flow. The flow must have compiled cleanly. Returns GraphModeOut.

deactivate()

client.graph.deactivate(study_id) -> GraphModeOut

Switch back to linear mode. The stored flow is kept. Returns GraphModeOut.

get_signals()

client.graph.get_signals(study_id) -> FlowSignalsOut

The delivery config of every “Send a signal” (webhook) step in the saved flow, including each step’s signing secret. Verify the deliveries with deutero.webhooks.unwrap(). Returns FlowSignalsOut.

describe_node_types()

client.graph.describe_node_types() -> dict[str, Any]

The configuration schema for every step type. Read this before authoring a flow. Returns the raw JSON schema as a dict.

Client

Study setup

Recruitment

Simulation

Results

Organization

Utilities

get()

set()

patch()

check()

import_questions()

activate()

deactivate()

get_signals()

describe_node_types()

​get()

​set()

​patch()

​check()

​import_questions()

​activate()

​deactivate()

​get_signals()

​describe_node_types()

get()

set()

patch()

check()

import_questions()

activate()

deactivate()

get_signals()

describe_node_types()