Skip to content

Validation rules + predictors

The CRUD flip side of dhis2 maintenance validation run + dhis2 maintenance predictors run. Both domains now expose authoring surfaces alongside the run endpoints:

Accessor API path Purpose
client.validation_rules /api/validationRules Compare two DHIS2 expressions (leftSide vs rightSide) under a chosen operator; fire violations when the comparison fails.
client.validation_rule_groups /api/validationRuleGroups Collect rules into named runs for maintenance validation run --group.
client.predictors /api/predictors CRUD + run. Generates synthetic data values from an expression over historical samples.
client.predictor_groups /api/predictorGroups Collect predictors into named runs for maintenance predictors run --group.

Expression sides are nested objects

ValidationRule.leftSide / rightSide and Predictor.generator are all DHIS2 Expression sub-objects. Each carries at least an expression string + missingValueStrategy flag. The accessors assemble the wrapper from plain kwargs so callers hand in the expression text directly:

rule = await client.validation_rules.create(
    name="BCG doses > 0",
    short_name="BCGgt0",
    left_expression="#{deBCG000001}",
    operator=Operator.GREATER_THAN,
    right_expression="0",
    importance=Importance.HIGH,
    organisation_unit_levels=[4],  # facility level
)

predictor = await client.predictors.create(
    name="BCG 3-month rolling average",
    short_name="BCG3mAvg",
    expression="#{deBCG000001}",
    output_data_element_uid="deOutput0001",
    sequential_sample_count=3,
    organisation_unit_level_uids=["ouLvlFac001"],
)

organisationUnitLevels asymmetry (upstream quirk)

Both models carry organisationUnitLevels but with different shapes:

  • ValidationRule.organisationUnitLevels — list of integer level numbers ([4]).
  • Predictor.organisationUnitLevels — list of OrganisationUnitLevel references ([{"id": uid}]).

The accessors expose this as organisation_unit_levels: list[int] vs organisation_unit_level_uids: list[str] respectively, matching DHIS2's wire shape. DHIS2 v42 returns a 500 on predictor create without valid level references.

missingValueStrategy

Defaults to SKIP_IF_ALL_VALUES_MISSING on both sides — rows where every operand is null are excluded from the comparison instead of counting as a violation. Flip to NEVER_SKIP to fail any row missing an operand.

No *Spec builder

Same call as every other authoring accessor: keyword args. Continues the spec-audit data point.

CLI

# ValidationRule + group
dhis2 metadata validation-rules create \
    --name "BCG gt zero" --short-name BCGgt0 \
    --left "#{deBCG000001}" --operator greater_than --right "0" \
    --importance HIGH --ou-level 4
dhis2 metadata validation-rule-groups create --name "BCG rules"
dhis2 metadata validation-rule-groups add-members <GRP_UID> --rule <RULE_UID>

# Predictor + group
dhis2 metadata predictors create \
    --name "BCG 3m avg" --short-name BCG3m \
    --expression "#{deBCG000001}" --output deOutput0001 \
    --sequential 3 --ou-level ouLvlFac001
dhis2 metadata predictor-groups create --name "BCG predictors"
dhis2 metadata predictor-groups add-members <PDG_UID> --predictor <PRD_UID>

Every list has an ls alias; every destructive verb accepts --yes / -y.

MCP

24 tools mirroring the CLI: metadata_validation_rule_* (list / get / create / rename / delete), metadata_validation_rule_group_* (list / get / members / create / add-members / remove-members / delete), metadata_predictor_* (list / get / create / rename / delete), metadata_predictor_group_* (list / get / members / create / add-members / remove-members / delete).

Running them

Creating the rule or predictor is decoupled from running it:

  • dhis2 maintenance validation run --group <GRP_UID> --ds <DATASET_UID> --start-date … --end-date …
  • dhis2 maintenance predictors run --group <PDG_UID> --start-date … --end-date …

See the maintenance plugin for the run-side reference.

validation_rules

ValidationRule authoring — Dhis2Client.validation_rules.

ValidationRules compare two DHIS2 expressions (leftSide vs rightSide) against a configurable operator and fire violations when the comparison fails for a given period + organisation unit. They drive dhis2 maintenance validation run; DHIS2 ships dozens of built-in rules for aggregate data-quality checks.

This module adds the authoring primitives — run lives on Dhis2Client.validation.run_analysis(...):

  • create(...) — named kwargs over the minimal required subset (name, short_name, left_expression, operator, right_expression) plus common knobs (period_type, importance, missing_value_strategy, description).
  • update(rule) / rename(uid, ...) — standard edit pathways.
  • delete(uid) — drops the rule and any outstanding results.

leftSide / rightSide are Expression sub-objects on the wire; we build them from the string + strategy here so callers don't have to assemble the payload manually.

No *Spec builder — continues the spec-audit data point.

Classes

ValidationRule

Bases: BaseModel

Generated model for DHIS2 ValidationRule.

DHIS2 Validation Rule - persisted metadata (generated from /api/schemas at DHIS2 v42).

API endpoint: /api/validationRules.

Field Field(description=...) entries flag DHIS2 semantics the bare type can't capture: which side of a relationship owns the link (writable) vs the inverse side (ignored by the API), uniqueness constraints, and length bounds.

Source code in packages/dhis2w-client/src/dhis2w_client/generated/v42/schemas/validation_rule.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
class ValidationRule(BaseModel):
    """Generated model for DHIS2 `ValidationRule`.

    DHIS2 Validation Rule - persisted metadata (generated from /api/schemas at DHIS2 v42).

    API endpoint: /api/validationRules.

    Field `Field(description=...)` entries flag DHIS2 semantics the bare
    type can't capture: which side of a relationship owns the link
    (writable) vs the inverse side (ignored by the API), uniqueness
    constraints, and length bounds.
    """

    model_config = ConfigDict(extra="allow", populate_by_name=True)

    access: Any | None = Field(default=None, description="Reference to Access. Read-only (inverse side).")
    aggregateExportAttributeOptionCombo: str | None = Field(default=None, description="Length/value max=2147483647.")
    aggregateExportCategoryOptionCombo: str | None = Field(default=None, description="Length/value max=2147483647.")
    aggregationType: AggregationType | None = None
    attributeValues: Any | None = Field(default=None, description="Reference to AttributeValues. Length/value max=255.")
    code: str | None = Field(default=None, description="Unique. Length/value max=50.")
    created: datetime | None = None
    createdBy: Reference | None = Field(default=None, description="Reference to User.")
    description: str | None = Field(default=None, description="Length/value min=2, max=2147483647.")
    dimensionItem: str | None = Field(default=None, description="Read-only.")
    dimensionItemType: DimensionItemType | None = None
    displayDescription: str | None = Field(default=None, description="Read-only.")
    displayFormName: str | None = Field(default=None, description="Read-only.")
    displayInstruction: str | None = Field(default=None, description="Read-only.")
    displayName: str | None = Field(default=None, description="Read-only.")
    displayShortName: str | None = Field(default=None, description="Read-only.")
    favorite: bool | None = Field(default=None, description="Read-only.")
    favorites: list[Any] | None = Field(default=None, description="Collection of String. Read-only (inverse side).")
    formName: str | None = Field(default=None, description="Length/value max=2147483647.")
    href: str | None = None
    id: str | None = Field(default=None, description="Unique. Length/value min=11, max=11.")
    importance: Importance | None = None
    instruction: str | None = Field(default=None, description="Length/value max=2147483647.")
    lastUpdated: datetime | None = None
    lastUpdatedBy: Reference | None = Field(default=None, description="Reference to User.")
    leftSide: Any | None = Field(default=None, description="Reference to Expression. Unique. Length/value max=255.")
    legendSet: Reference | None = Field(default=None, description="Reference to LegendSet. Read-only (inverse side).")
    legendSets: list[Any] | None = Field(default=None, description="Collection of LegendSet. Read-only (inverse side).")
    name: str | None = Field(default=None, description="Unique. Length/value min=1, max=230.")
    notificationTemplates: list[Any] | None = Field(
        default=None, description="Collection of ValidationNotificationTemplate. Read-only (inverse side)."
    )
    operator: Operator | None = None
    organisationUnitLevels: list[Any] | None = Field(default=None, description="Collection of Integer.")
    periodType: PeriodType | None = Field(default=None, description="Reference to PeriodType. Length/value max=255.")
    queryMods: Any | None = Field(default=None, description="Reference to QueryModifiers. Read-only (inverse side).")
    rightSide: Any | None = Field(default=None, description="Reference to Expression. Unique. Length/value max=255.")
    sharing: Any | None = Field(default=None, description="Reference to Sharing. Length/value max=255.")
    shortName: str | None = Field(default=None, description="Unique. Length/value min=1, max=50.")
    skipFormValidation: bool | None = None
    translations: list[Any] | None = Field(default=None, description="Collection of Translation. Length/value max=255.")
    user: Reference | None = Field(default=None, description="Reference to User. Read-only (inverse side).")
    validationRuleGroups: list[Any] | None = Field(
        default=None, description="Collection of ValidationRuleGroup. Read-only (inverse side)."
    )

ValidationRulesAccessor

Dhis2Client.validation_rules — CRUD helpers over /api/validationRules.

Source code in packages/dhis2w-client/src/dhis2w_client/validation_rules.py 
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
class ValidationRulesAccessor:
    """`Dhis2Client.validation_rules` — CRUD helpers over `/api/validationRules`."""

    def __init__(self, client: Dhis2Client) -> None:
        """Bind to the sharing client."""
        self._client = client

    async def list_all(
        self,
        *,
        period_type: PeriodType | str | None = None,
        page: int = 1,
        page_size: int = 50,
    ) -> list[ValidationRule]:
        """Page through ValidationRules, optionally filtered by periodType."""
        filters: list[str] | None = None
        if period_type is not None:
            value = period_type.value if isinstance(period_type, PeriodType) else period_type
            filters = [f"periodType:eq:{value}"]
        return cast(
            list[ValidationRule],
            await self._client.resources.validation_rules.list(
                fields=_VR_FIELDS,
                filters=filters,
                page=page,
                page_size=page_size,
            ),
        )

    async def get(self, uid: str) -> ValidationRule:
        """Fetch one ValidationRule with both expression sides + group refs inline."""
        return await self._client.get(
            f"/api/validationRules/{uid}", model=ValidationRule, params={"fields": _VR_FIELDS}
        )

    async def create(
        self,
        *,
        name: str,
        short_name: str,
        left_expression: str,
        operator: Operator | str,
        right_expression: str,
        period_type: PeriodType | str = PeriodType.MONTHLY,
        importance: Importance | str = Importance.MEDIUM,
        missing_value_strategy: MissingValueStrategy | str = MissingValueStrategy.SKIP_IF_ALL_VALUES_MISSING,
        left_description: str | None = None,
        right_description: str | None = None,
        description: str | None = None,
        code: str | None = None,
        organisation_unit_levels: list[int] | None = None,
        uid: str | None = None,
    ) -> ValidationRule:
        """Create a ValidationRule.

        `left_expression` / `right_expression` use DHIS2's aggregate
        expression syntax (`#{<DE_UID>.<CC_UID>}` for disaggregated
        DEs, `#{<DE_UID>}` for default-combo DEs). `operator` picks the
        comparison — `EQUAL_TO`, `LESS_THAN_OR_EQUAL_TO`, etc. The
        `missing_value_strategy` default skips rows where every ref is
        null so blank cells don't count as a violation.

        `organisation_unit_levels` scopes the rule to specific depths —
        pass `[4]` to restrict it to facility-level OUs.
        """
        payload: dict[str, Any] = {
            "name": name,
            "shortName": short_name,
            "periodType": period_type.value if isinstance(period_type, PeriodType) else period_type,
            "importance": importance.value if isinstance(importance, Importance) else importance,
            "operator": operator.value if isinstance(operator, Operator) else operator,
            "leftSide": _expression(
                expression=left_expression,
                description=left_description or (description and f"leftSide: {description}") or None,
                strategy=missing_value_strategy,
            ),
            "rightSide": _expression(
                expression=right_expression,
                description=right_description or (description and f"rightSide: {description}") or None,
                strategy=missing_value_strategy,
            ),
        }
        if uid:
            payload["id"] = uid
        if code:
            payload["code"] = code
        if description:
            payload["description"] = description
        if organisation_unit_levels:
            payload["organisationUnitLevels"] = list(organisation_unit_levels)
        envelope = await self._client.post("/api/validationRules", payload, model=WebMessageResponse)
        created_uid = envelope.created_uid or uid
        if not created_uid:
            raise RuntimeError("validation-rule create did not return a uid")
        return await self.get(created_uid)

    async def update(self, rule: ValidationRule) -> ValidationRule:
        """PUT an edited ValidationRule back. `rule.id` must be set."""
        if not rule.id:
            raise ValueError("update requires rule.id to be set")
        body = rule.model_dump(by_alias=True, exclude_none=True, mode="json")
        await self._client.put_raw(f"/api/validationRules/{rule.id}", body=body)
        return await self.get(rule.id)

    async def rename(
        self,
        uid: str,
        *,
        name: str | None = None,
        short_name: str | None = None,
        description: str | None = None,
    ) -> ValidationRule:
        """Partial-update shortcut — read, mutate the label fields, PUT."""
        if name is None and short_name is None and description is None:
            raise ValueError("rename requires at least one of name / short_name / description")
        current = await self.get(uid)
        if name is not None:
            current.name = name
        if short_name is not None:
            current.shortName = short_name
        if description is not None:
            current.description = description
        return await self.update(current)

    async def delete(self, uid: str) -> None:
        """Delete a ValidationRule — DHIS2 removes any outstanding results it had raised."""
        if not uid:
            raise ValueError("delete requires a non-empty uid")
        await self._client.resources.validation_rules.delete(uid)
Functions
__init__(client)

Bind to the sharing client.

Source code in packages/dhis2w-client/src/dhis2w_client/validation_rules.py 
51
52
53
def __init__(self, client: Dhis2Client) -> None:
    """Bind to the sharing client."""
    self._client = client
list_all(*, period_type=None, page=1, page_size=50) async

Page through ValidationRules, optionally filtered by periodType.

Source code in packages/dhis2w-client/src/dhis2w_client/validation_rules.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
async def list_all(
    self,
    *,
    period_type: PeriodType | str | None = None,
    page: int = 1,
    page_size: int = 50,
) -> list[ValidationRule]:
    """Page through ValidationRules, optionally filtered by periodType."""
    filters: list[str] | None = None
    if period_type is not None:
        value = period_type.value if isinstance(period_type, PeriodType) else period_type
        filters = [f"periodType:eq:{value}"]
    return cast(
        list[ValidationRule],
        await self._client.resources.validation_rules.list(
            fields=_VR_FIELDS,
            filters=filters,
            page=page,
            page_size=page_size,
        ),
    )
get(uid) async

Fetch one ValidationRule with both expression sides + group refs inline.

Source code in packages/dhis2w-client/src/dhis2w_client/validation_rules.py 
77
78
79
80
81
async def get(self, uid: str) -> ValidationRule:
    """Fetch one ValidationRule with both expression sides + group refs inline."""
    return await self._client.get(
        f"/api/validationRules/{uid}", model=ValidationRule, params={"fields": _VR_FIELDS}
    )
create(*, name, short_name, left_expression, operator, right_expression, period_type=PeriodType.MONTHLY, importance=Importance.MEDIUM, missing_value_strategy=MissingValueStrategy.SKIP_IF_ALL_VALUES_MISSING, left_description=None, right_description=None, description=None, code=None, organisation_unit_levels=None, uid=None) async

Create a ValidationRule.

left_expression / right_expression use DHIS2's aggregate expression syntax (#{<DE_UID>.<CC_UID>} for disaggregated DEs, #{<DE_UID>} for default-combo DEs). operator picks the comparison — EQUAL_TO, LESS_THAN_OR_EQUAL_TO, etc. The missing_value_strategy default skips rows where every ref is null so blank cells don't count as a violation.

organisation_unit_levels scopes the rule to specific depths — pass [4] to restrict it to facility-level OUs.

Source code in packages/dhis2w-client/src/dhis2w_client/validation_rules.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
async def create(
    self,
    *,
    name: str,
    short_name: str,
    left_expression: str,
    operator: Operator | str,
    right_expression: str,
    period_type: PeriodType | str = PeriodType.MONTHLY,
    importance: Importance | str = Importance.MEDIUM,
    missing_value_strategy: MissingValueStrategy | str = MissingValueStrategy.SKIP_IF_ALL_VALUES_MISSING,
    left_description: str | None = None,
    right_description: str | None = None,
    description: str | None = None,
    code: str | None = None,
    organisation_unit_levels: list[int] | None = None,
    uid: str | None = None,
) -> ValidationRule:
    """Create a ValidationRule.

    `left_expression` / `right_expression` use DHIS2's aggregate
    expression syntax (`#{<DE_UID>.<CC_UID>}` for disaggregated
    DEs, `#{<DE_UID>}` for default-combo DEs). `operator` picks the
    comparison — `EQUAL_TO`, `LESS_THAN_OR_EQUAL_TO`, etc. The
    `missing_value_strategy` default skips rows where every ref is
    null so blank cells don't count as a violation.

    `organisation_unit_levels` scopes the rule to specific depths —
    pass `[4]` to restrict it to facility-level OUs.
    """
    payload: dict[str, Any] = {
        "name": name,
        "shortName": short_name,
        "periodType": period_type.value if isinstance(period_type, PeriodType) else period_type,
        "importance": importance.value if isinstance(importance, Importance) else importance,
        "operator": operator.value if isinstance(operator, Operator) else operator,
        "leftSide": _expression(
            expression=left_expression,
            description=left_description or (description and f"leftSide: {description}") or None,
            strategy=missing_value_strategy,
        ),
        "rightSide": _expression(
            expression=right_expression,
            description=right_description or (description and f"rightSide: {description}") or None,
            strategy=missing_value_strategy,
        ),
    }
    if uid:
        payload["id"] = uid
    if code:
        payload["code"] = code
    if description:
        payload["description"] = description
    if organisation_unit_levels:
        payload["organisationUnitLevels"] = list(organisation_unit_levels)
    envelope = await self._client.post("/api/validationRules", payload, model=WebMessageResponse)
    created_uid = envelope.created_uid or uid
    if not created_uid:
        raise RuntimeError("validation-rule create did not return a uid")
    return await self.get(created_uid)
update(rule) async

PUT an edited ValidationRule back. rule.id must be set.

Source code in packages/dhis2w-client/src/dhis2w_client/validation_rules.py 
144
145
146
147
148
149
150
async def update(self, rule: ValidationRule) -> ValidationRule:
    """PUT an edited ValidationRule back. `rule.id` must be set."""
    if not rule.id:
        raise ValueError("update requires rule.id to be set")
    body = rule.model_dump(by_alias=True, exclude_none=True, mode="json")
    await self._client.put_raw(f"/api/validationRules/{rule.id}", body=body)
    return await self.get(rule.id)
rename(uid, *, name=None, short_name=None, description=None) async

Partial-update shortcut — read, mutate the label fields, PUT.

Source code in packages/dhis2w-client/src/dhis2w_client/validation_rules.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
async def rename(
    self,
    uid: str,
    *,
    name: str | None = None,
    short_name: str | None = None,
    description: str | None = None,
) -> ValidationRule:
    """Partial-update shortcut — read, mutate the label fields, PUT."""
    if name is None and short_name is None and description is None:
        raise ValueError("rename requires at least one of name / short_name / description")
    current = await self.get(uid)
    if name is not None:
        current.name = name
    if short_name is not None:
        current.shortName = short_name
    if description is not None:
        current.description = description
    return await self.update(current)
delete(uid) async

Delete a ValidationRule — DHIS2 removes any outstanding results it had raised.

Source code in packages/dhis2w-client/src/dhis2w_client/validation_rules.py 
172
173
174
175
176
async def delete(self, uid: str) -> None:
    """Delete a ValidationRule — DHIS2 removes any outstanding results it had raised."""
    if not uid:
        raise ValueError("delete requires a non-empty uid")
    await self._client.resources.validation_rules.delete(uid)

validation_rule_groups

ValidationRuleGroup authoring — Dhis2Client.validation_rule_groups.

ValidationRuleGroups collect ValidationRules into named runs so dhis2 maintenance validation run --group <uid> exercises a coherent subset (BCG-dose rules, ANC rules, …). Follows the same CRUD + per-item membership pattern as IndicatorGroupsAccessor.

Classes

ValidationRuleGroup

Bases: BaseModel

Generated model for DHIS2 ValidationRuleGroup.

DHIS2 Validation Rule Group - persisted metadata (generated from /api/schemas at DHIS2 v42).

API endpoint: /api/validationRuleGroups.

Field Field(description=...) entries flag DHIS2 semantics the bare type can't capture: which side of a relationship owns the link (writable) vs the inverse side (ignored by the API), uniqueness constraints, and length bounds.

Source code in packages/dhis2w-client/src/dhis2w_client/generated/v42/schemas/validation_rule_group.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class ValidationRuleGroup(BaseModel):
    """Generated model for DHIS2 `ValidationRuleGroup`.

    DHIS2 Validation Rule Group - persisted metadata (generated from /api/schemas at DHIS2 v42).

    API endpoint: /api/validationRuleGroups.

    Field `Field(description=...)` entries flag DHIS2 semantics the bare
    type can't capture: which side of a relationship owns the link
    (writable) vs the inverse side (ignored by the API), uniqueness
    constraints, and length bounds.
    """

    model_config = ConfigDict(extra="allow", populate_by_name=True)

    access: Any | None = Field(default=None, description="Reference to Access. Read-only (inverse side).")
    attributeValues: Any | None = Field(default=None, description="Reference to AttributeValues. Length/value max=255.")
    code: str | None = Field(default=None, description="Unique. Length/value max=50.")
    created: datetime | None = None
    createdBy: Reference | None = Field(default=None, description="Reference to User.")
    description: str | None = Field(default=None, description="Length/value min=2, max=2147483647.")
    displayName: str | None = Field(default=None, description="Read-only.")
    favorite: bool | None = Field(default=None, description="Read-only.")
    favorites: list[Any] | None = Field(default=None, description="Collection of String. Read-only (inverse side).")
    href: str | None = None
    id: str | None = Field(default=None, description="Unique. Length/value min=11, max=11.")
    lastUpdated: datetime | None = None
    lastUpdatedBy: Reference | None = Field(default=None, description="Reference to User.")
    name: str | None = Field(default=None, description="Unique. Length/value min=1, max=230.")
    sharing: Any | None = Field(default=None, description="Reference to Sharing. Length/value max=255.")
    translations: list[Any] | None = Field(default=None, description="Collection of Translation. Length/value max=255.")
    user: Reference | None = Field(default=None, description="Reference to User. Read-only (inverse side).")
    validationRules: list[Any] | None = Field(default=None, description="Collection of ValidationRule.")

ValidationRuleGroupsAccessor

Dhis2Client.validation_rule_groups — CRUD + membership over /api/validationRuleGroups.

Source code in packages/dhis2w-client/src/dhis2w_client/validation_rule_groups.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
class ValidationRuleGroupsAccessor:
    """`Dhis2Client.validation_rule_groups` — CRUD + membership over `/api/validationRuleGroups`."""

    def __init__(self, client: Dhis2Client) -> None:
        """Bind to the sharing client."""
        self._client = client

    async def list_all(self) -> list[ValidationRuleGroup]:
        """Return every ValidationRuleGroup."""
        return cast(
            list[ValidationRuleGroup],
            await self._client.resources.validation_rule_groups.list(
                fields=_GROUP_FIELDS,
                paging=False,
            ),
        )

    async def get(self, uid: str) -> ValidationRuleGroup:
        """Fetch one group by UID with its `validationRules` refs populated."""
        return await self._client.get(
            f"/api/validationRuleGroups/{uid}", model=ValidationRuleGroup, params={"fields": _GROUP_FIELDS}
        )

    async def list_members(
        self,
        uid: str,
        *,
        page: int = 1,
        page_size: int = 50,
    ) -> list[ValidationRule]:
        """Page through ValidationRules belonging to one group."""
        return cast(
            list[ValidationRule],
            await self._client.resources.validation_rules.list(
                fields=_MEMBER_FIELDS,
                filters=[f"validationRuleGroups.id:eq:{uid}"],
                order=["name:asc"],
                page=page,
                page_size=page_size,
            ),
        )

    async def create(
        self,
        *,
        name: str,
        short_name: str | None = None,
        uid: str | None = None,
        code: str | None = None,
        description: str | None = None,
    ) -> ValidationRuleGroup:
        """Create an empty ValidationRuleGroup; add members afterwards via `add_members`."""
        payload: dict[str, Any] = {"name": name}
        if short_name:
            payload["shortName"] = short_name
        if uid:
            payload["id"] = uid
        if code:
            payload["code"] = code
        if description:
            payload["description"] = description
        envelope = await self._client.post("/api/validationRuleGroups", payload, model=WebMessageResponse)
        created_uid = envelope.created_uid or uid
        if not created_uid:
            raise RuntimeError("validation-rule-group create did not return a uid")
        return await self.get(created_uid)

    async def update(self, group: ValidationRuleGroup) -> ValidationRuleGroup:
        """PUT an edited ValidationRuleGroup back. `group.id` must be set."""
        if not group.id:
            raise ValueError("update requires group.id to be set")
        body = group.model_dump(by_alias=True, exclude_none=True, mode="json")
        await self._client.put_raw(f"/api/validationRuleGroups/{group.id}", body=body)
        return await self.get(group.id)

    async def add_members(self, uid: str, *, validation_rule_uids: list[str]) -> ValidationRuleGroup:
        """Add ValidationRules to the group via the per-item POST shortcut."""
        for rule_uid in validation_rule_uids:
            await self._client.resources.validation_rule_groups.add_collection_item(uid, "validationRules", rule_uid)
        return await self.get(uid)

    async def remove_members(self, uid: str, *, validation_rule_uids: list[str]) -> ValidationRuleGroup:
        """Drop ValidationRules from the group via the per-item DELETE shortcut."""
        for rule_uid in validation_rule_uids:
            await self._client.resources.validation_rule_groups.remove_collection_item(uid, "validationRules", rule_uid)
        return await self.get(uid)

    async def delete(self, uid: str) -> None:
        """Delete the grouping row — member rules stay."""
        if not uid:
            raise ValueError("delete requires a non-empty uid")
        await self._client.resources.validation_rule_groups.delete(uid)
Functions
__init__(client)

Bind to the sharing client.

Source code in packages/dhis2w-client/src/dhis2w_client/validation_rule_groups.py 
26
27
28
def __init__(self, client: Dhis2Client) -> None:
    """Bind to the sharing client."""
    self._client = client
list_all() async

Return every ValidationRuleGroup.

Source code in packages/dhis2w-client/src/dhis2w_client/validation_rule_groups.py 
30
31
32
33
34
35
36
37
38
async def list_all(self) -> list[ValidationRuleGroup]:
    """Return every ValidationRuleGroup."""
    return cast(
        list[ValidationRuleGroup],
        await self._client.resources.validation_rule_groups.list(
            fields=_GROUP_FIELDS,
            paging=False,
        ),
    )
get(uid) async

Fetch one group by UID with its validationRules refs populated.

Source code in packages/dhis2w-client/src/dhis2w_client/validation_rule_groups.py 
40
41
42
43
44
async def get(self, uid: str) -> ValidationRuleGroup:
    """Fetch one group by UID with its `validationRules` refs populated."""
    return await self._client.get(
        f"/api/validationRuleGroups/{uid}", model=ValidationRuleGroup, params={"fields": _GROUP_FIELDS}
    )
list_members(uid, *, page=1, page_size=50) async

Page through ValidationRules belonging to one group.

Source code in packages/dhis2w-client/src/dhis2w_client/validation_rule_groups.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
async def list_members(
    self,
    uid: str,
    *,
    page: int = 1,
    page_size: int = 50,
) -> list[ValidationRule]:
    """Page through ValidationRules belonging to one group."""
    return cast(
        list[ValidationRule],
        await self._client.resources.validation_rules.list(
            fields=_MEMBER_FIELDS,
            filters=[f"validationRuleGroups.id:eq:{uid}"],
            order=["name:asc"],
            page=page,
            page_size=page_size,
        ),
    )
create(*, name, short_name=None, uid=None, code=None, description=None) async

Create an empty ValidationRuleGroup; add members afterwards via add_members.

Source code in packages/dhis2w-client/src/dhis2w_client/validation_rule_groups.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
async def create(
    self,
    *,
    name: str,
    short_name: str | None = None,
    uid: str | None = None,
    code: str | None = None,
    description: str | None = None,
) -> ValidationRuleGroup:
    """Create an empty ValidationRuleGroup; add members afterwards via `add_members`."""
    payload: dict[str, Any] = {"name": name}
    if short_name:
        payload["shortName"] = short_name
    if uid:
        payload["id"] = uid
    if code:
        payload["code"] = code
    if description:
        payload["description"] = description
    envelope = await self._client.post("/api/validationRuleGroups", payload, model=WebMessageResponse)
    created_uid = envelope.created_uid or uid
    if not created_uid:
        raise RuntimeError("validation-rule-group create did not return a uid")
    return await self.get(created_uid)
update(group) async

PUT an edited ValidationRuleGroup back. group.id must be set.

Source code in packages/dhis2w-client/src/dhis2w_client/validation_rule_groups.py 
90
91
92
93
94
95
96
async def update(self, group: ValidationRuleGroup) -> ValidationRuleGroup:
    """PUT an edited ValidationRuleGroup back. `group.id` must be set."""
    if not group.id:
        raise ValueError("update requires group.id to be set")
    body = group.model_dump(by_alias=True, exclude_none=True, mode="json")
    await self._client.put_raw(f"/api/validationRuleGroups/{group.id}", body=body)
    return await self.get(group.id)
add_members(uid, *, validation_rule_uids) async

Add ValidationRules to the group via the per-item POST shortcut.

Source code in packages/dhis2w-client/src/dhis2w_client/validation_rule_groups.py 
 98
 99
100
101
102
async def add_members(self, uid: str, *, validation_rule_uids: list[str]) -> ValidationRuleGroup:
    """Add ValidationRules to the group via the per-item POST shortcut."""
    for rule_uid in validation_rule_uids:
        await self._client.resources.validation_rule_groups.add_collection_item(uid, "validationRules", rule_uid)
    return await self.get(uid)
remove_members(uid, *, validation_rule_uids) async

Drop ValidationRules from the group via the per-item DELETE shortcut.

Source code in packages/dhis2w-client/src/dhis2w_client/validation_rule_groups.py 
104
105
106
107
108
async def remove_members(self, uid: str, *, validation_rule_uids: list[str]) -> ValidationRuleGroup:
    """Drop ValidationRules from the group via the per-item DELETE shortcut."""
    for rule_uid in validation_rule_uids:
        await self._client.resources.validation_rule_groups.remove_collection_item(uid, "validationRules", rule_uid)
    return await self.get(uid)
delete(uid) async

Delete the grouping row — member rules stay.

Source code in packages/dhis2w-client/src/dhis2w_client/validation_rule_groups.py 
110
111
112
113
114
async def delete(self, uid: str) -> None:
    """Delete the grouping row — member rules stay."""
    if not uid:
        raise ValueError("delete requires a non-empty uid")
    await self._client.resources.validation_rule_groups.delete(uid)

predictors

Predictor authoring + run — Dhis2Client.predictors.

Predictors generate data values from expressions over historical data (e.g. "3-month rolling average of X" → emit a synthetic DataElement row). The accessor covers both authoring (create / update / delete) and the run endpoints DHIS2 exposes:

  • POST /api/predictors/run?startDate=…&endDate=… — run every predictor on the instance.
  • POST /api/predictors/{uid}/run?startDate=…&endDate=… — run one.
  • POST /api/predictorGroups/{uid}/run?startDate=…&endDate=… — run a named group of predictors in one pass (exposed from PredictorsAccessor.run_group for backward compatibility + also from PredictorGroupsAccessor.run).

All three run shapes return a WebMessageResponse with a summary of predictions written / ignored / failed; none of them kick a background job, so there's no task to watch.

Authoring surface: Predictor.generator is an Expression sub-object typed as Any on the generated schema. create(...) assembles the minimal wrapper here so callers pass the expression string + description, not the nested payload.

No *Spec builder — continues the spec-audit data point.

Classes

Predictor

Bases: BaseModel

Generated model for DHIS2 Predictor.

DHIS2 Predictor - persisted metadata (generated from /api/schemas at DHIS2 v42).

API endpoint: /api/predictors.

Field Field(description=...) entries flag DHIS2 semantics the bare type can't capture: which side of a relationship owns the link (writable) vs the inverse side (ignored by the API), uniqueness constraints, and length bounds.

Source code in packages/dhis2w-client/src/dhis2w_client/generated/v42/schemas/predictor.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
class Predictor(BaseModel):
    """Generated model for DHIS2 `Predictor`.

    DHIS2 Predictor - persisted metadata (generated from /api/schemas at DHIS2 v42).

    API endpoint: /api/predictors.

    Field `Field(description=...)` entries flag DHIS2 semantics the bare
    type can't capture: which side of a relationship owns the link
    (writable) vs the inverse side (ignored by the API), uniqueness
    constraints, and length bounds.
    """

    model_config = ConfigDict(extra="allow", populate_by_name=True)

    access: Any | None = Field(default=None, description="Reference to Access. Read-only (inverse side).")
    annualSampleCount: int | None = Field(default=None, description="Length/value max=10.")
    attributeValues: Any | None = Field(
        default=None, description="Reference to AttributeValues. Read-only (inverse side)."
    )
    code: str | None = Field(default=None, description="Unique. Length/value max=50.")
    created: datetime | None = None
    createdBy: Reference | None = Field(default=None, description="Reference to User. Read-only (inverse side).")
    description: str | None = Field(default=None, description="Length/value min=1, max=2147483647.")
    displayDescription: str | None = Field(default=None, description="Read-only.")
    displayFormName: str | None = Field(default=None, description="Read-only.")
    displayName: str | None = Field(default=None, description="Read-only.")
    displayShortName: str | None = Field(default=None, description="Read-only.")
    favorite: bool | None = Field(default=None, description="Read-only.")
    favorites: list[Any] | None = Field(default=None, description="Collection of String. Read-only (inverse side).")
    formName: str | None = Field(default=None, description="Length/value max=2147483647.")
    generator: Any | None = Field(default=None, description="Reference to Expression. Unique. Length/value max=255.")
    href: str | None = None
    id: str | None = Field(default=None, description="Unique. Length/value min=11, max=11.")
    lastUpdated: datetime | None = None
    lastUpdatedBy: Reference | None = Field(default=None, description="Reference to User.")
    name: str | None = Field(default=None, description="Unique. Length/value min=1, max=230.")
    organisationUnitDescendants: OrganisationUnitDescendants | None = None
    organisationUnitLevels: list[Any] | None = Field(default=None, description="Collection of OrganisationUnitLevel.")
    output: Reference | None = Field(default=None, description="Reference to DataElement.")
    outputCombo: Reference | None = Field(default=None, description="Reference to CategoryOptionCombo.")
    periodType: PeriodType | None = Field(default=None, description="Reference to PeriodType. Length/value max=255.")
    predictorGroups: list[Any] | None = Field(
        default=None, description="Collection of PredictorGroup. Read-only (inverse side)."
    )
    sampleSkipTest: Any | None = Field(
        default=None, description="Reference to Expression. Unique. Length/value max=255."
    )
    sequentialSampleCount: int | None = Field(default=None, description="Length/value max=2147483647.")
    sequentialSkipCount: int | None = Field(default=None, description="Length/value max=2147483647.")
    sharing: Any | None = Field(default=None, description="Reference to Sharing. Read-only (inverse side).")
    shortName: str | None = Field(default=None, description="Unique. Length/value min=1, max=50.")
    translations: list[Any] | None = Field(default=None, description="Collection of Translation. Length/value max=255.")
    user: Reference | None = Field(default=None, description="Reference to User. Read-only (inverse side).")

PredictorsAccessor

Dhis2Client.predictors — CRUD + run helpers over /api/predictors.

Source code in packages/dhis2w-client/src/dhis2w_client/predictors.py 
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
class PredictorsAccessor:
    """`Dhis2Client.predictors` — CRUD + run helpers over `/api/predictors`."""

    def __init__(self, client: Dhis2Client) -> None:
        """Bind to the sharing client."""
        self._client = client

    # ---- CRUD -----------------------------------------------------------

    async def list_all(
        self,
        *,
        period_type: PeriodType | str | None = None,
        page: int = 1,
        page_size: int = 50,
    ) -> list[Predictor]:
        """Page through Predictors, optionally filtered by periodType."""
        filters: list[str] | None = None
        if period_type is not None:
            value = period_type.value if isinstance(period_type, PeriodType) else period_type
            filters = [f"periodType:eq:{value}"]
        return cast(
            list[Predictor],
            await self._client.resources.predictors.list(
                fields=_PREDICTOR_FIELDS,
                filters=filters,
                page=page,
                page_size=page_size,
            ),
        )

    async def get(self, uid: str) -> Predictor:
        """Fetch one Predictor with generator, output, OU scope resolved inline."""
        return await self._client.get(f"/api/predictors/{uid}", model=Predictor, params={"fields": _PREDICTOR_FIELDS})

    async def create(
        self,
        *,
        name: str,
        short_name: str,
        expression: str,
        output_data_element_uid: str,
        period_type: PeriodType | str = PeriodType.MONTHLY,
        sequential_sample_count: int = 3,
        annual_sample_count: int = 0,
        sequential_skip_count: int = 0,
        organisation_unit_descendants: OrganisationUnitDescendants | str = OrganisationUnitDescendants.SELECTED,
        organisation_unit_level_uids: list[str] | None = None,
        output_combo_uid: str | None = None,
        missing_value_strategy: MissingValueStrategy | str = MissingValueStrategy.SKIP_IF_ALL_VALUES_MISSING,
        generator_description: str | None = None,
        description: str | None = None,
        code: str | None = None,
        uid: str | None = None,
    ) -> Predictor:
        """Create a Predictor.

        `expression` uses DHIS2's aggregate expression syntax; the
        accessor wraps it in the `generator` Expression sub-object.
        `output_data_element_uid` is the target DE the prediction writes
        to — needs a `TRACKER` or `AGGREGATE` domain and a numeric
        valueType.

        `sequential_sample_count` + `annual_sample_count` control the
        look-back window: `3` monthly samples with the default period
        type averages the three prior months.

        `organisation_unit_level_uids` scopes the run — pass the UIDs of
        the `OrganisationUnitLevel` rows the predictor should cover
        (typically the facility level for data-entry predictors).
        """
        payload: dict[str, Any] = {
            "name": name,
            "shortName": short_name,
            "periodType": period_type.value if isinstance(period_type, PeriodType) else period_type,
            "sequentialSampleCount": sequential_sample_count,
            "annualSampleCount": annual_sample_count,
            "sequentialSkipCount": sequential_skip_count,
            "organisationUnitDescendants": (
                organisation_unit_descendants.value
                if isinstance(organisation_unit_descendants, OrganisationUnitDescendants)
                else organisation_unit_descendants
            ),
            "output": {"id": output_data_element_uid},
            "generator": {
                "expression": expression,
                "missingValueStrategy": (
                    missing_value_strategy.value
                    if isinstance(missing_value_strategy, MissingValueStrategy)
                    else missing_value_strategy
                ),
                "slidingWindow": False,
            },
        }
        if generator_description:
            payload["generator"]["description"] = generator_description
        if output_combo_uid:
            payload["outputCombo"] = {"id": output_combo_uid}
        if organisation_unit_level_uids:
            payload["organisationUnitLevels"] = [{"id": level_uid} for level_uid in organisation_unit_level_uids]
        if uid:
            payload["id"] = uid
        if code:
            payload["code"] = code
        if description:
            payload["description"] = description
        envelope = await self._client.post("/api/predictors", payload, model=WebMessageResponse)
        created_uid = envelope.created_uid or uid
        if not created_uid:
            raise RuntimeError("predictor create did not return a uid")
        return await self.get(created_uid)

    async def update(self, predictor: Predictor) -> Predictor:
        """PUT an edited Predictor back. `predictor.id` must be set."""
        if not predictor.id:
            raise ValueError("update requires predictor.id to be set")
        body = predictor.model_dump(by_alias=True, exclude_none=True, mode="json")
        await self._client.put_raw(f"/api/predictors/{predictor.id}", body=body)
        return await self.get(predictor.id)

    async def rename(
        self,
        uid: str,
        *,
        name: str | None = None,
        short_name: str | None = None,
        description: str | None = None,
    ) -> Predictor:
        """Partial-update shortcut — read, mutate the label fields, PUT."""
        if name is None and short_name is None and description is None:
            raise ValueError("rename requires at least one of name / short_name / description")
        current = await self.get(uid)
        if name is not None:
            current.name = name
        if short_name is not None:
            current.shortName = short_name
        if description is not None:
            current.description = description
        return await self.update(current)

    async def delete(self, uid: str) -> None:
        """Delete a Predictor. DHIS2 keeps any data values it already wrote."""
        if not uid:
            raise ValueError("delete requires a non-empty uid")
        await self._client.resources.predictors.delete(uid)

    # ---- Run ------------------------------------------------------------

    async def run_all(self, *, start_date: str, end_date: str) -> WebMessageResponse:
        """Run every predictor on the instance for the given date range.

        Returns the summary envelope — `.import_count()` gives
        `imported / updated / ignored / deleted` counts for the emitted
        data values.
        """
        return await self._run("/api/predictors/run", start_date=start_date, end_date=end_date)

    async def run_one(self, predictor_uid: str, *, start_date: str, end_date: str) -> WebMessageResponse:
        """Run a single predictor by UID over the given date range."""
        return await self._run(
            f"/api/predictors/{predictor_uid}/run",
            start_date=start_date,
            end_date=end_date,
        )

    async def run_group(self, group_uid: str, *, start_date: str, end_date: str) -> WebMessageResponse:
        """Run every predictor in a `PredictorGroup` over the given date range."""
        return await self._run(
            f"/api/predictorGroups/{group_uid}/run",
            start_date=start_date,
            end_date=end_date,
        )

    async def _run(self, path: str, *, start_date: str, end_date: str) -> WebMessageResponse:
        """Dispatch a predictor-run POST + return the typed envelope."""
        params: dict[str, Any] = {"startDate": start_date, "endDate": end_date}
        return await self._client.post(path, body=None, params=params, model=WebMessageResponse)
Functions
__init__(client)

Bind to the sharing client.

Source code in packages/dhis2w-client/src/dhis2w_client/predictors.py 
54
55
56
def __init__(self, client: Dhis2Client) -> None:
    """Bind to the sharing client."""
    self._client = client
list_all(*, period_type=None, page=1, page_size=50) async

Page through Predictors, optionally filtered by periodType.

Source code in packages/dhis2w-client/src/dhis2w_client/predictors.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
async def list_all(
    self,
    *,
    period_type: PeriodType | str | None = None,
    page: int = 1,
    page_size: int = 50,
) -> list[Predictor]:
    """Page through Predictors, optionally filtered by periodType."""
    filters: list[str] | None = None
    if period_type is not None:
        value = period_type.value if isinstance(period_type, PeriodType) else period_type
        filters = [f"periodType:eq:{value}"]
    return cast(
        list[Predictor],
        await self._client.resources.predictors.list(
            fields=_PREDICTOR_FIELDS,
            filters=filters,
            page=page,
            page_size=page_size,
        ),
    )
get(uid) async

Fetch one Predictor with generator, output, OU scope resolved inline.

Source code in packages/dhis2w-client/src/dhis2w_client/predictors.py 
82
83
84
async def get(self, uid: str) -> Predictor:
    """Fetch one Predictor with generator, output, OU scope resolved inline."""
    return await self._client.get(f"/api/predictors/{uid}", model=Predictor, params={"fields": _PREDICTOR_FIELDS})
create(*, name, short_name, expression, output_data_element_uid, period_type=PeriodType.MONTHLY, sequential_sample_count=3, annual_sample_count=0, sequential_skip_count=0, organisation_unit_descendants=OrganisationUnitDescendants.SELECTED, organisation_unit_level_uids=None, output_combo_uid=None, missing_value_strategy=MissingValueStrategy.SKIP_IF_ALL_VALUES_MISSING, generator_description=None, description=None, code=None, uid=None) async

Create a Predictor.

expression uses DHIS2's aggregate expression syntax; the accessor wraps it in the generator Expression sub-object. output_data_element_uid is the target DE the prediction writes to — needs a TRACKER or AGGREGATE domain and a numeric valueType.

sequential_sample_count + annual_sample_count control the look-back window: 3 monthly samples with the default period type averages the three prior months.

organisation_unit_level_uids scopes the run — pass the UIDs of the OrganisationUnitLevel rows the predictor should cover (typically the facility level for data-entry predictors).

Source code in packages/dhis2w-client/src/dhis2w_client/predictors.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
async def create(
    self,
    *,
    name: str,
    short_name: str,
    expression: str,
    output_data_element_uid: str,
    period_type: PeriodType | str = PeriodType.MONTHLY,
    sequential_sample_count: int = 3,
    annual_sample_count: int = 0,
    sequential_skip_count: int = 0,
    organisation_unit_descendants: OrganisationUnitDescendants | str = OrganisationUnitDescendants.SELECTED,
    organisation_unit_level_uids: list[str] | None = None,
    output_combo_uid: str | None = None,
    missing_value_strategy: MissingValueStrategy | str = MissingValueStrategy.SKIP_IF_ALL_VALUES_MISSING,
    generator_description: str | None = None,
    description: str | None = None,
    code: str | None = None,
    uid: str | None = None,
) -> Predictor:
    """Create a Predictor.

    `expression` uses DHIS2's aggregate expression syntax; the
    accessor wraps it in the `generator` Expression sub-object.
    `output_data_element_uid` is the target DE the prediction writes
    to — needs a `TRACKER` or `AGGREGATE` domain and a numeric
    valueType.

    `sequential_sample_count` + `annual_sample_count` control the
    look-back window: `3` monthly samples with the default period
    type averages the three prior months.

    `organisation_unit_level_uids` scopes the run — pass the UIDs of
    the `OrganisationUnitLevel` rows the predictor should cover
    (typically the facility level for data-entry predictors).
    """
    payload: dict[str, Any] = {
        "name": name,
        "shortName": short_name,
        "periodType": period_type.value if isinstance(period_type, PeriodType) else period_type,
        "sequentialSampleCount": sequential_sample_count,
        "annualSampleCount": annual_sample_count,
        "sequentialSkipCount": sequential_skip_count,
        "organisationUnitDescendants": (
            organisation_unit_descendants.value
            if isinstance(organisation_unit_descendants, OrganisationUnitDescendants)
            else organisation_unit_descendants
        ),
        "output": {"id": output_data_element_uid},
        "generator": {
            "expression": expression,
            "missingValueStrategy": (
                missing_value_strategy.value
                if isinstance(missing_value_strategy, MissingValueStrategy)
                else missing_value_strategy
            ),
            "slidingWindow": False,
        },
    }
    if generator_description:
        payload["generator"]["description"] = generator_description
    if output_combo_uid:
        payload["outputCombo"] = {"id": output_combo_uid}
    if organisation_unit_level_uids:
        payload["organisationUnitLevels"] = [{"id": level_uid} for level_uid in organisation_unit_level_uids]
    if uid:
        payload["id"] = uid
    if code:
        payload["code"] = code
    if description:
        payload["description"] = description
    envelope = await self._client.post("/api/predictors", payload, model=WebMessageResponse)
    created_uid = envelope.created_uid or uid
    if not created_uid:
        raise RuntimeError("predictor create did not return a uid")
    return await self.get(created_uid)
update(predictor) async

PUT an edited Predictor back. predictor.id must be set.

Source code in packages/dhis2w-client/src/dhis2w_client/predictors.py 
163
164
165
166
167
168
169
async def update(self, predictor: Predictor) -> Predictor:
    """PUT an edited Predictor back. `predictor.id` must be set."""
    if not predictor.id:
        raise ValueError("update requires predictor.id to be set")
    body = predictor.model_dump(by_alias=True, exclude_none=True, mode="json")
    await self._client.put_raw(f"/api/predictors/{predictor.id}", body=body)
    return await self.get(predictor.id)
rename(uid, *, name=None, short_name=None, description=None) async

Partial-update shortcut — read, mutate the label fields, PUT.

Source code in packages/dhis2w-client/src/dhis2w_client/predictors.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
async def rename(
    self,
    uid: str,
    *,
    name: str | None = None,
    short_name: str | None = None,
    description: str | None = None,
) -> Predictor:
    """Partial-update shortcut — read, mutate the label fields, PUT."""
    if name is None and short_name is None and description is None:
        raise ValueError("rename requires at least one of name / short_name / description")
    current = await self.get(uid)
    if name is not None:
        current.name = name
    if short_name is not None:
        current.shortName = short_name
    if description is not None:
        current.description = description
    return await self.update(current)
delete(uid) async

Delete a Predictor. DHIS2 keeps any data values it already wrote.

Source code in packages/dhis2w-client/src/dhis2w_client/predictors.py 
191
192
193
194
195
async def delete(self, uid: str) -> None:
    """Delete a Predictor. DHIS2 keeps any data values it already wrote."""
    if not uid:
        raise ValueError("delete requires a non-empty uid")
    await self._client.resources.predictors.delete(uid)
run_all(*, start_date, end_date) async

Run every predictor on the instance for the given date range.

Returns the summary envelope — .import_count() gives imported / updated / ignored / deleted counts for the emitted data values.

Source code in packages/dhis2w-client/src/dhis2w_client/predictors.py 
199
200
201
202
203
204
205
206
async def run_all(self, *, start_date: str, end_date: str) -> WebMessageResponse:
    """Run every predictor on the instance for the given date range.

    Returns the summary envelope — `.import_count()` gives
    `imported / updated / ignored / deleted` counts for the emitted
    data values.
    """
    return await self._run("/api/predictors/run", start_date=start_date, end_date=end_date)
run_one(predictor_uid, *, start_date, end_date) async

Run a single predictor by UID over the given date range.

Source code in packages/dhis2w-client/src/dhis2w_client/predictors.py 
208
209
210
211
212
213
214
async def run_one(self, predictor_uid: str, *, start_date: str, end_date: str) -> WebMessageResponse:
    """Run a single predictor by UID over the given date range."""
    return await self._run(
        f"/api/predictors/{predictor_uid}/run",
        start_date=start_date,
        end_date=end_date,
    )
run_group(group_uid, *, start_date, end_date) async

Run every predictor in a PredictorGroup over the given date range.

Source code in packages/dhis2w-client/src/dhis2w_client/predictors.py 
216
217
218
219
220
221
222
async def run_group(self, group_uid: str, *, start_date: str, end_date: str) -> WebMessageResponse:
    """Run every predictor in a `PredictorGroup` over the given date range."""
    return await self._run(
        f"/api/predictorGroups/{group_uid}/run",
        start_date=start_date,
        end_date=end_date,
    )

predictor_groups

PredictorGroup authoring — Dhis2Client.predictor_groups.

PredictorGroups collect Predictors so dhis2 maintenance predictors run --group <uid> exercises a coherent subset in one pass. Mirrors the IndicatorGroup / ValidationRuleGroup CRUD + per-item membership pattern.

Running a group lives on PredictorsAccessor.run_group (kept for backward compatibility with existing callers); the group accessor focuses on the authoring verbs.

Classes

PredictorGroup

Bases: BaseModel

Generated model for DHIS2 PredictorGroup.

DHIS2 Predictor Group - persisted metadata (generated from /api/schemas at DHIS2 v42).

API endpoint: /api/predictorGroups.

Field Field(description=...) entries flag DHIS2 semantics the bare type can't capture: which side of a relationship owns the link (writable) vs the inverse side (ignored by the API), uniqueness constraints, and length bounds.

Source code in packages/dhis2w-client/src/dhis2w_client/generated/v42/schemas/predictor_group.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class PredictorGroup(BaseModel):
    """Generated model for DHIS2 `PredictorGroup`.

    DHIS2 Predictor Group - persisted metadata (generated from /api/schemas at DHIS2 v42).

    API endpoint: /api/predictorGroups.

    Field `Field(description=...)` entries flag DHIS2 semantics the bare
    type can't capture: which side of a relationship owns the link
    (writable) vs the inverse side (ignored by the API), uniqueness
    constraints, and length bounds.
    """

    model_config = ConfigDict(extra="allow", populate_by_name=True)

    access: Any | None = Field(default=None, description="Reference to Access. Read-only (inverse side).")
    attributeValues: Any | None = Field(
        default=None, description="Reference to AttributeValues. Read-only (inverse side)."
    )
    code: str | None = Field(default=None, description="Unique. Length/value max=50.")
    created: datetime | None = None
    createdBy: Reference | None = Field(default=None, description="Reference to User.")
    description: str | None = Field(default=None, description="Length/value min=2, max=2147483647.")
    displayName: str | None = Field(default=None, description="Read-only.")
    favorite: bool | None = Field(default=None, description="Read-only.")
    favorites: list[Any] | None = Field(default=None, description="Collection of String. Read-only (inverse side).")
    href: str | None = None
    id: str | None = Field(default=None, description="Unique. Length/value min=11, max=11.")
    lastUpdated: datetime | None = None
    lastUpdatedBy: Reference | None = Field(default=None, description="Reference to User.")
    name: str | None = Field(default=None, description="Unique. Length/value min=1, max=230.")
    predictors: list[Any] | None = Field(default=None, description="Collection of Predictor.")
    sharing: Any | None = Field(default=None, description="Reference to Sharing. Length/value max=255.")
    translations: list[Any] | None = Field(default=None, description="Collection of Translation. Length/value max=255.")
    user: Reference | None = Field(default=None, description="Reference to User. Read-only (inverse side).")

PredictorGroupsAccessor

Dhis2Client.predictor_groups — CRUD + membership over /api/predictorGroups.

Source code in packages/dhis2w-client/src/dhis2w_client/predictor_groups.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
class PredictorGroupsAccessor:
    """`Dhis2Client.predictor_groups` — CRUD + membership over `/api/predictorGroups`."""

    def __init__(self, client: Dhis2Client) -> None:
        """Bind to the sharing client."""
        self._client = client

    async def list_all(self) -> list[PredictorGroup]:
        """Return every PredictorGroup."""
        return cast(
            list[PredictorGroup],
            await self._client.resources.predictor_groups.list(
                fields=_GROUP_FIELDS,
                paging=False,
            ),
        )

    async def get(self, uid: str) -> PredictorGroup:
        """Fetch one group by UID with its `predictors` refs populated."""
        return await self._client.get(
            f"/api/predictorGroups/{uid}", model=PredictorGroup, params={"fields": _GROUP_FIELDS}
        )

    async def list_members(
        self,
        uid: str,
        *,
        page: int = 1,
        page_size: int = 50,
    ) -> list[Predictor]:
        """Page through Predictors belonging to one group."""
        return cast(
            list[Predictor],
            await self._client.resources.predictors.list(
                fields=_MEMBER_FIELDS,
                filters=[f"predictorGroups.id:eq:{uid}"],
                order=["name:asc"],
                page=page,
                page_size=page_size,
            ),
        )

    async def create(
        self,
        *,
        name: str,
        short_name: str | None = None,
        uid: str | None = None,
        code: str | None = None,
        description: str | None = None,
    ) -> PredictorGroup:
        """Create an empty PredictorGroup; add members afterwards via `add_members`."""
        payload: dict[str, Any] = {"name": name}
        if short_name:
            payload["shortName"] = short_name
        if uid:
            payload["id"] = uid
        if code:
            payload["code"] = code
        if description:
            payload["description"] = description
        envelope = await self._client.post("/api/predictorGroups", payload, model=WebMessageResponse)
        created_uid = envelope.created_uid or uid
        if not created_uid:
            raise RuntimeError("predictor-group create did not return a uid")
        return await self.get(created_uid)

    async def update(self, group: PredictorGroup) -> PredictorGroup:
        """PUT an edited PredictorGroup back. `group.id` must be set."""
        if not group.id:
            raise ValueError("update requires group.id to be set")
        body = group.model_dump(by_alias=True, exclude_none=True, mode="json")
        await self._client.put_raw(f"/api/predictorGroups/{group.id}", body=body)
        return await self.get(group.id)

    async def add_members(self, uid: str, *, predictor_uids: list[str]) -> PredictorGroup:
        """Add Predictors to the group via the per-item POST shortcut."""
        for predictor_uid in predictor_uids:
            await self._client.resources.predictor_groups.add_collection_item(uid, "predictors", predictor_uid)
        return await self.get(uid)

    async def remove_members(self, uid: str, *, predictor_uids: list[str]) -> PredictorGroup:
        """Drop Predictors from the group via the per-item DELETE shortcut."""
        for predictor_uid in predictor_uids:
            await self._client.resources.predictor_groups.remove_collection_item(uid, "predictors", predictor_uid)
        return await self.get(uid)

    async def delete(self, uid: str) -> None:
        """Delete the grouping row — member predictors stay."""
        if not uid:
            raise ValueError("delete requires a non-empty uid")
        await self._client.resources.predictor_groups.delete(uid)
Functions
__init__(client)

Bind to the sharing client.

Source code in packages/dhis2w-client/src/dhis2w_client/predictor_groups.py 
30
31
32
def __init__(self, client: Dhis2Client) -> None:
    """Bind to the sharing client."""
    self._client = client
list_all() async

Return every PredictorGroup.

Source code in packages/dhis2w-client/src/dhis2w_client/predictor_groups.py 
34
35
36
37
38
39
40
41
42
async def list_all(self) -> list[PredictorGroup]:
    """Return every PredictorGroup."""
    return cast(
        list[PredictorGroup],
        await self._client.resources.predictor_groups.list(
            fields=_GROUP_FIELDS,
            paging=False,
        ),
    )
get(uid) async

Fetch one group by UID with its predictors refs populated.

Source code in packages/dhis2w-client/src/dhis2w_client/predictor_groups.py 
44
45
46
47
48
async def get(self, uid: str) -> PredictorGroup:
    """Fetch one group by UID with its `predictors` refs populated."""
    return await self._client.get(
        f"/api/predictorGroups/{uid}", model=PredictorGroup, params={"fields": _GROUP_FIELDS}
    )
list_members(uid, *, page=1, page_size=50) async

Page through Predictors belonging to one group.

Source code in packages/dhis2w-client/src/dhis2w_client/predictor_groups.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
async def list_members(
    self,
    uid: str,
    *,
    page: int = 1,
    page_size: int = 50,
) -> list[Predictor]:
    """Page through Predictors belonging to one group."""
    return cast(
        list[Predictor],
        await self._client.resources.predictors.list(
            fields=_MEMBER_FIELDS,
            filters=[f"predictorGroups.id:eq:{uid}"],
            order=["name:asc"],
            page=page,
            page_size=page_size,
        ),
    )
create(*, name, short_name=None, uid=None, code=None, description=None) async

Create an empty PredictorGroup; add members afterwards via add_members.

Source code in packages/dhis2w-client/src/dhis2w_client/predictor_groups.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
async def create(
    self,
    *,
    name: str,
    short_name: str | None = None,
    uid: str | None = None,
    code: str | None = None,
    description: str | None = None,
) -> PredictorGroup:
    """Create an empty PredictorGroup; add members afterwards via `add_members`."""
    payload: dict[str, Any] = {"name": name}
    if short_name:
        payload["shortName"] = short_name
    if uid:
        payload["id"] = uid
    if code:
        payload["code"] = code
    if description:
        payload["description"] = description
    envelope = await self._client.post("/api/predictorGroups", payload, model=WebMessageResponse)
    created_uid = envelope.created_uid or uid
    if not created_uid:
        raise RuntimeError("predictor-group create did not return a uid")
    return await self.get(created_uid)
update(group) async

PUT an edited PredictorGroup back. group.id must be set.

Source code in packages/dhis2w-client/src/dhis2w_client/predictor_groups.py 
 94
 95
 96
 97
 98
 99
100
async def update(self, group: PredictorGroup) -> PredictorGroup:
    """PUT an edited PredictorGroup back. `group.id` must be set."""
    if not group.id:
        raise ValueError("update requires group.id to be set")
    body = group.model_dump(by_alias=True, exclude_none=True, mode="json")
    await self._client.put_raw(f"/api/predictorGroups/{group.id}", body=body)
    return await self.get(group.id)
add_members(uid, *, predictor_uids) async

Add Predictors to the group via the per-item POST shortcut.

Source code in packages/dhis2w-client/src/dhis2w_client/predictor_groups.py 
102
103
104
105
106
async def add_members(self, uid: str, *, predictor_uids: list[str]) -> PredictorGroup:
    """Add Predictors to the group via the per-item POST shortcut."""
    for predictor_uid in predictor_uids:
        await self._client.resources.predictor_groups.add_collection_item(uid, "predictors", predictor_uid)
    return await self.get(uid)
remove_members(uid, *, predictor_uids) async

Drop Predictors from the group via the per-item DELETE shortcut.

Source code in packages/dhis2w-client/src/dhis2w_client/predictor_groups.py 
108
109
110
111
112
async def remove_members(self, uid: str, *, predictor_uids: list[str]) -> PredictorGroup:
    """Drop Predictors from the group via the per-item DELETE shortcut."""
    for predictor_uid in predictor_uids:
        await self._client.resources.predictor_groups.remove_collection_item(uid, "predictors", predictor_uid)
    return await self.get(uid)
delete(uid) async

Delete the grouping row — member predictors stay.

Source code in packages/dhis2w-client/src/dhis2w_client/predictor_groups.py 
114
115
116
117
118
async def delete(self, uid: str) -> None:
    """Delete the grouping row — member predictors stay."""
    if not uid:
        raise ValueError("delete requires a non-empty uid")
    await self._client.resources.predictor_groups.delete(uid)