Skip to content

Maintenance

MaintenanceAccessor on Dhis2Client.maintenance — the data-integrity reader (get_integrity_report, iter_integrity_issues) plus the on-demand CategoryOptionCombo matrix regeneration (update_category_option_combos). DHIS2's other background-job triggers (analytics refresh, predictor runs, validation runs, cache clears) live in the matching plugin services and on the CLI / MCP — see Triggering analytics / monitoring refresh below for the right entry point.

When to reach for it

  • Run DHIS2's built-in data-integrity scan (81 checks) and pull the typed report.
  • Stream tagged integrity issues as they're emitted (large instances can have thousands; iter_integrity_issues is the streaming consumer).
  • Trigger COC matrix regeneration after a CategoryCombo save on v43 (update_category_option_combos) — see also category_combos.wait_for_coc_generation for the polling helper that pairs with it.

Worked example — stream integrity issues

from dhis2w_core.client_context import open_client
from dhis2w_core.profile import profile_from_env

async with open_client(profile_from_env()) as client:
    # `iter_integrity_issues` is an async iterator — one `IntegrityIssueRow`
    # per row as DHIS2 emits it; safe for instances with thousands of issues.
    # Each row carries the owning check's metadata (`check_name`,
    # `check_display_name`, `severity`) plus the typed issue itself
    # (`row.issue.name`, `row.issue.id`, `row.issue.comment`).
    severity_counts: dict[str, int] = {}
    async for row in client.maintenance.iter_integrity_issues():
        sev = row.severity or "UNKNOWN"
        severity_counts[sev] = severity_counts.get(sev, 0) + 1
        if severity_counts[sev] <= 3:
            print(f"  [{sev}] {row.check_display_name or row.check_name}: {row.issue.name} ({row.issue.id})")
    print(f"total by severity: {severity_counts}")

Worked example — full integrity report (snapshot, not stream)

async with open_client(profile_from_env()) as client:
    # Returns a typed `DataIntegrityReport` carrying `.results`, a
    # `dict[str, DataIntegrityResult]` keyed by check name. Each result
    # has `.name`, `.severity`, `.count`, and `.issues` (list of
    # DataIntegrityIssue with `.name`, `.id`, `.comment`).
    report = await client.maintenance.get_integrity_report()
    print(f"{len(report.results)} checks ran")
    for check_key, result in list(report.results.items())[:5]:
        print(f"  {check_key}  severity={result.severity}  issues={len(result.issues)}")

Triggering analytics / monitoring refresh

The accessor doesn't expose a refresh trigger directly — refresh is a plugin-service surface, not a raw-client one. Three real paths:

  1. CLI: d2w maintenance refresh analytics --watch.
  2. MCP: maintenance_refresh_analytics tool with watch=true.
  3. Python via plugin service: import dhis2w_core.v42.plugins.maintenance.service.refresh_analytics(profile, ...) directly — see examples/v42/client/task_polling.py for the full kick-off + poll-with-client.tasks.await_completion pattern.

maintenance

Typed models + client accessor for DHIS2 maintenance + data-integrity + task-notification APIs.

DataIntegrityCheck and DataIntegrityIssue come from dhis2w_client.generated.v42.oas. DataIntegrityResult and DataIntegrityReport stay hand-written — OpenAPI splits the result into separate DataIntegrityDetails / DataIntegritySummary shapes, but this module's callers want the merged view + the client-side {check_name: result} map. Notification re-exports the OAS type so callers get typed category: JobType, dataType: NotificationDataType, level: NotificationLevel enums + time: datetime.

MaintenanceAccessor (bound to Dhis2Client.maintenance) exposes the data-integrity read paths. iter_integrity_issues is the ergonomic entry point for large runs — yields one issue at a time tagged with its owning check's metadata, so callers don't have to walk the {check_name: {issues: [...]}} two-level shape themselves.

Classes

DataIntegrityCheck

Bases: BaseModel

OpenAPI schema DataIntegrityCheck.

Source code in packages/dhis2w-client/src/dhis2w_client/generated/v42/oas/data_integrity_check.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
class DataIntegrityCheck(_BaseModel):
    """OpenAPI schema `DataIntegrityCheck`."""

    model_config = _ConfigDict(extra="allow", populate_by_name=True, defer_build=True)

    averageExecutionTime: int | None = None
    code: str | None = None
    description: str | None = None
    displayName: str | None = None
    introduction: str | None = None
    isProgrammatic: bool | None = None
    isSlow: bool | None = None
    issuesIdType: str | None = None
    name: str | None = None
    recommendation: str | None = None
    section: str | None = None
    sectionOrder: int | None = None
    severity: DataIntegritySeverity | None = None

DataIntegrityIssue

Bases: BaseModel

OpenAPI schema DataIntegrityIssue.

Source code in packages/dhis2w-client/src/dhis2w_client/generated/v42/oas/data_integrity_issue.py 
10
11
12
13
14
15
16
17
18
class DataIntegrityIssue(_BaseModel):
    """OpenAPI schema `DataIntegrityIssue`."""

    model_config = _ConfigDict(extra="allow", populate_by_name=True, defer_build=True)

    comment: str | None = None
    id: str | None = None
    name: str | None = None
    refs: list[str] | None = None

Notification

Bases: BaseModel

OpenAPI schema Notification.

Source code in packages/dhis2w-client/src/dhis2w_client/generated/v42/oas/notification.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
class Notification(_BaseModel):
    """OpenAPI schema `Notification`."""

    model_config = _ConfigDict(extra="allow", populate_by_name=True, defer_build=True)

    category: JobType | None = None
    completed: bool | None = None
    data: Any | None = None
    dataType: NotificationDataType | None = None
    id: str | None = None
    level: NotificationLevel | None = None
    message: str | None = None
    time: datetime | None = None
    uid: str | None = None

JobType

Bases: StrEnum

JobType.

Source code in packages/dhis2w-client/src/dhis2w_client/generated/v42/oas/_enums.py 
2097
2098
2099
2100
2101
2102
2103
2104
2105
2106
2107
2108
2109
2110
2111
2112
2113
2114
2115
2116
2117
2118
2119
2120
2121
2122
2123
2124
2125
2126
2127
2128
2129
2130
2131
2132
2133
2134
2135
2136
2137
2138
2139
2140
2141
2142
2143
class JobType(StrEnum):
    """JobType."""

    DATA_INTEGRITY = "DATA_INTEGRITY"
    DATA_INTEGRITY_DETAILS = "DATA_INTEGRITY_DETAILS"
    RESOURCE_TABLE = "RESOURCE_TABLE"
    ANALYTICS_TABLE = "ANALYTICS_TABLE"
    CONTINUOUS_ANALYTICS_TABLE = "CONTINUOUS_ANALYTICS_TABLE"
    SINGLE_EVENT_DATA_SYNC = "SINGLE_EVENT_DATA_SYNC"
    TRACKED_ENTITY_DATA_SYNC = "TRACKED_ENTITY_DATA_SYNC"
    DATA_SYNC = "DATA_SYNC"
    META_DATA_SYNC = "META_DATA_SYNC"
    AGGREGATE_DATA_EXCHANGE = "AGGREGATE_DATA_EXCHANGE"
    SEND_SCHEDULED_MESSAGE = "SEND_SCHEDULED_MESSAGE"
    PROGRAM_NOTIFICATIONS = "PROGRAM_NOTIFICATIONS"
    MONITORING = "MONITORING"
    PUSH_ANALYSIS = "PUSH_ANALYSIS"
    HTML_PUSH_ANALYTICS = "HTML_PUSH_ANALYTICS"
    TRACKER_SEARCH_OPTIMIZATION = "TRACKER_SEARCH_OPTIMIZATION"
    PREDICTOR = "PREDICTOR"
    MATERIALIZED_SQL_VIEW_UPDATE = "MATERIALIZED_SQL_VIEW_UPDATE"
    DISABLE_INACTIVE_USERS = "DISABLE_INACTIVE_USERS"
    TEST = "TEST"
    LOCK_EXCEPTION_CLEANUP = "LOCK_EXCEPTION_CLEANUP"
    MOCK = "MOCK"
    SMS_SEND = "SMS_SEND"
    SMS_INBOUND_PROCESSING = "SMS_INBOUND_PROCESSING"
    TRACKER_IMPORT_JOB = "TRACKER_IMPORT_JOB"
    TRACKER_IMPORT_NOTIFICATION_JOB = "TRACKER_IMPORT_NOTIFICATION_JOB"
    TRACKER_IMPORT_RULE_ENGINE_JOB = "TRACKER_IMPORT_RULE_ENGINE_JOB"
    IMAGE_PROCESSING = "IMAGE_PROCESSING"
    COMPLETE_DATA_SET_REGISTRATION_IMPORT = "COMPLETE_DATA_SET_REGISTRATION_IMPORT"
    DATAVALUE_IMPORT_INTERNAL = "DATAVALUE_IMPORT_INTERNAL"
    METADATA_IMPORT = "METADATA_IMPORT"
    DATAVALUE_IMPORT = "DATAVALUE_IMPORT"
    GEOJSON_IMPORT = "GEOJSON_IMPORT"
    GML_IMPORT = "GML_IMPORT"
    HOUSEKEEPING = "HOUSEKEEPING"
    DATA_VALUE_TRIM = "DATA_VALUE_TRIM"
    DATA_SET_NOTIFICATION = "DATA_SET_NOTIFICATION"
    CREDENTIALS_EXPIRY_ALERT = "CREDENTIALS_EXPIRY_ALERT"
    DATA_STATISTICS = "DATA_STATISTICS"
    FILE_RESOURCE_CLEANUP = "FILE_RESOURCE_CLEANUP"
    ACCOUNT_EXPIRY_ALERT = "ACCOUNT_EXPIRY_ALERT"
    VALIDATION_RESULTS_NOTIFICATION = "VALIDATION_RESULTS_NOTIFICATION"
    REMOVE_USED_OR_EXPIRED_RESERVED_VALUES = "REMOVE_USED_OR_EXPIRED_RESERVED_VALUES"
    SYSTEM_VERSION_UPDATE_CHECK = "SYSTEM_VERSION_UPDATE_CHECK"

NotificationDataType

Bases: StrEnum

NotificationDataType.

Source code in packages/dhis2w-client/src/dhis2w_client/generated/v42/oas/_enums.py 
2490
2491
2492
2493
class NotificationDataType(StrEnum):
    """NotificationDataType."""

    PARAMETERS = "PARAMETERS"

NotificationLevel

Bases: StrEnum

NotificationLevel.

Source code in packages/dhis2w-client/src/dhis2w_client/generated/v42/oas/_enums.py 
2496
2497
2498
2499
2500
2501
2502
2503
2504
class NotificationLevel(StrEnum):
    """NotificationLevel."""

    OFF = "OFF"
    DEBUG = "DEBUG"
    LOOP = "LOOP"
    INFO = "INFO"
    WARN = "WARN"
    ERROR = "ERROR"

DataIntegrityResult

Bases: BaseModel

Result of one check — populated after the async job completes.

Merges OpenAPI's DataIntegrityDetails (has issues[]) and DataIntegritySummary (has count) into one caller-friendly shape. Both modes populate startTime / finishedTime once the job has run; an unrun check returns the definition block alone.

Source code in packages/dhis2w-client/src/dhis2w_client/v42/maintenance.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
class DataIntegrityResult(BaseModel):
    """Result of one check — populated after the async job completes.

    Merges OpenAPI's `DataIntegrityDetails` (has `issues[]`) and
    `DataIntegritySummary` (has `count`) into one caller-friendly shape.
    Both modes populate `startTime` / `finishedTime` once the job has run;
    an unrun check returns the definition block alone.
    """

    model_config = ConfigDict(extra="allow")

    name: str
    displayName: str | None = None
    section: str | None = None
    severity: str | None = None
    code: str | None = None
    count: int | None = None
    issues: list[DataIntegrityIssue] = Field(default_factory=list)
    startTime: str | None = None
    finishedTime: str | None = None
    averageExecutionTime: int | None = None

DataIntegrityReport

Bases: BaseModel

/api/dataIntegrity/summary or /details response — keyed by check name.

Hand-written: DHIS2 returns {check_name: result} — a client-side convenience shape not in OpenAPI. The from_api classmethod hides the raw-dict detail.

Source code in packages/dhis2w-client/src/dhis2w_client/v42/maintenance.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
class DataIntegrityReport(BaseModel):
    """`/api/dataIntegrity/summary` or `/details` response — keyed by check name.

    Hand-written: DHIS2 returns `{check_name: result}` — a client-side convenience
    shape not in OpenAPI. The `from_api` classmethod hides the raw-dict detail.
    """

    model_config = ConfigDict(extra="allow")

    results: dict[str, DataIntegrityResult] = Field(default_factory=dict)

    @classmethod
    def from_api(cls, raw: dict[str, Any]) -> DataIntegrityReport:
        """Validate the raw `{check_name: {...}}` dict DHIS2 returns into a typed report."""
        results = {name: DataIntegrityResult.model_validate(body) for name, body in raw.items()}
        return cls(results=results)
Functions
from_api(raw) classmethod

Validate the raw {check_name: {...}} dict DHIS2 returns into a typed report.

Source code in packages/dhis2w-client/src/dhis2w_client/v42/maintenance.py 
67
68
69
70
71
@classmethod
def from_api(cls, raw: dict[str, Any]) -> DataIntegrityReport:
    """Validate the raw `{check_name: {...}}` dict DHIS2 returns into a typed report."""
    results = {name: DataIntegrityResult.model_validate(body) for name, body in raw.items()}
    return cls(results=results)

IntegrityIssueRow

Bases: BaseModel

One issue from a data-integrity run, tagged with its owning check's metadata.

iter_integrity_issues yields these as a flat stream so callers can filter / transform without walking the two-level {check_name: {issues: [...]}} shape themselves.

Source code in packages/dhis2w-client/src/dhis2w_client/v42/maintenance.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
class IntegrityIssueRow(BaseModel):
    """One issue from a data-integrity run, tagged with its owning check's metadata.

    `iter_integrity_issues` yields these as a flat stream so callers can
    filter / transform without walking the two-level
    `{check_name: {issues: [...]}}` shape themselves.
    """

    model_config = ConfigDict(frozen=True)

    check_name: str
    check_display_name: str | None = None
    severity: str | None = None
    issue: DataIntegrityIssue

MaintenanceAccessor

Dhis2Client.maintenance — read paths for the data-integrity surface.

Writes (kicking off a run, clearing cache) stay on the plugin-layer service in dhis2w_core for now — those need a Profile for OAuth2 token-store keying, which the raw client doesn't know about.

Source code in packages/dhis2w-client/src/dhis2w_client/v42/maintenance.py 
 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
class MaintenanceAccessor:
    """`Dhis2Client.maintenance` — read paths for the data-integrity surface.

    Writes (kicking off a run, clearing cache) stay on the plugin-layer
    service in `dhis2w_core` for now — those need a `Profile` for OAuth2
    token-store keying, which the raw client doesn't know about.
    """

    def __init__(self, client: Dhis2Client) -> None:
        """Bind to the sharing client — reuses its auth + HTTP pool for every request."""
        self._client = client

    async def get_integrity_report(
        self,
        *,
        checks: Sequence[str] | None = None,
        details: bool = True,
    ) -> DataIntegrityReport:
        """Fetch the full `/api/dataIntegrity/{details|summary}` report as a typed model.

        `details=True` (the default) populates `issues[]` on each result;
        `details=False` hits the cheaper `/summary` endpoint which returns
        just counts + timing. Pass `checks` to narrow to specific check
        names (from `list_dataintegrity_checks`); omit for every check
        the last run produced.
        """
        path = "/api/dataIntegrity/details" if details else "/api/dataIntegrity/summary"
        params: dict[str, list[str]] = {"checks": list(checks)} if checks else {}
        raw = await self._client.get_raw(path, params=params or None)
        return DataIntegrityReport.from_api(raw)

    async def iter_integrity_issues(
        self,
        *,
        checks: Sequence[str] | None = None,
    ) -> AsyncIterator[IntegrityIssueRow]:
        """Stream every issue from `/api/dataIntegrity/details` one at a time.

        DHIS2's endpoint returns the whole `{check_name: {issues: [...]}}`
        structure in one response (no server-side pagination). This helper
        still buys you:

        - A flat stream — `async for row in ...` instead of nested loops.
        - Tagged rows — each yielded `IntegrityIssueRow` carries the
          owning check's name + display name + severity, so the caller
          knows the provenance without a second lookup.
        - Early break — stop iteration mid-stream without building the
          full list in memory on the Python side.

        Issues yield in the order DHIS2 returns checks, then the order
        of that check's `issues[]` list — stable across runs.
        """
        report = await self.get_integrity_report(checks=checks, details=True)
        for check_name, result in report.results.items():
            for issue in result.issues:
                yield IntegrityIssueRow(
                    check_name=check_name,
                    check_display_name=result.displayName,
                    severity=result.severity,
                    issue=issue,
                )

    async def update_category_option_combos(self) -> None:
        """Trigger DHIS2 to (re)generate the CategoryOptionCombo matrix.

        DHIS2 v42 auto-generated COCs whenever a CategoryCombo was saved,
        so callers rarely needed this. v43 changed the behavior — saving
        a CategoryCombo no longer triggers regeneration; the matrix stays
        empty until this maintenance task runs.

        `client.category_combos.wait_for_coc_generation` calls this
        helper internally before polling so the helper "just works" on
        both versions. Call it directly when you need to ensure the COC
        matrix is up to date for an existing combo (e.g. after appending
        a category via `add_category`).

        The endpoint is synchronous — DHIS2 walks every persisted combo,
        adds missing COCs, removes orphaned ones, and returns when done.
        """
        await self._client.post_raw("/api/maintenance/categoryOptionComboUpdate", {})
Functions
__init__(client)

Bind to the sharing client — reuses its auth + HTTP pool for every request.

Source code in packages/dhis2w-client/src/dhis2w_client/v42/maintenance.py 
 98
 99
100
def __init__(self, client: Dhis2Client) -> None:
    """Bind to the sharing client — reuses its auth + HTTP pool for every request."""
    self._client = client
get_integrity_report(*, checks=None, details=True) async

Fetch the full /api/dataIntegrity/{details|summary} report as a typed model.

details=True (the default) populates issues[] on each result; details=False hits the cheaper /summary endpoint which returns just counts + timing. Pass checks to narrow to specific check names (from list_dataintegrity_checks); omit for every check the last run produced.

Source code in packages/dhis2w-client/src/dhis2w_client/v42/maintenance.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
async def get_integrity_report(
    self,
    *,
    checks: Sequence[str] | None = None,
    details: bool = True,
) -> DataIntegrityReport:
    """Fetch the full `/api/dataIntegrity/{details|summary}` report as a typed model.

    `details=True` (the default) populates `issues[]` on each result;
    `details=False` hits the cheaper `/summary` endpoint which returns
    just counts + timing. Pass `checks` to narrow to specific check
    names (from `list_dataintegrity_checks`); omit for every check
    the last run produced.
    """
    path = "/api/dataIntegrity/details" if details else "/api/dataIntegrity/summary"
    params: dict[str, list[str]] = {"checks": list(checks)} if checks else {}
    raw = await self._client.get_raw(path, params=params or None)
    return DataIntegrityReport.from_api(raw)
iter_integrity_issues(*, checks=None) async

Stream every issue from /api/dataIntegrity/details one at a time.

DHIS2's endpoint returns the whole {check_name: {issues: [...]}} structure in one response (no server-side pagination). This helper still buys you:

  • A flat stream — async for row in ... instead of nested loops.
  • Tagged rows — each yielded IntegrityIssueRow carries the owning check's name + display name + severity, so the caller knows the provenance without a second lookup.
  • Early break — stop iteration mid-stream without building the full list in memory on the Python side.

Issues yield in the order DHIS2 returns checks, then the order of that check's issues[] list — stable across runs.

Source code in packages/dhis2w-client/src/dhis2w_client/v42/maintenance.py 
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
async def iter_integrity_issues(
    self,
    *,
    checks: Sequence[str] | None = None,
) -> AsyncIterator[IntegrityIssueRow]:
    """Stream every issue from `/api/dataIntegrity/details` one at a time.

    DHIS2's endpoint returns the whole `{check_name: {issues: [...]}}`
    structure in one response (no server-side pagination). This helper
    still buys you:

    - A flat stream — `async for row in ...` instead of nested loops.
    - Tagged rows — each yielded `IntegrityIssueRow` carries the
      owning check's name + display name + severity, so the caller
      knows the provenance without a second lookup.
    - Early break — stop iteration mid-stream without building the
      full list in memory on the Python side.

    Issues yield in the order DHIS2 returns checks, then the order
    of that check's `issues[]` list — stable across runs.
    """
    report = await self.get_integrity_report(checks=checks, details=True)
    for check_name, result in report.results.items():
        for issue in result.issues:
            yield IntegrityIssueRow(
                check_name=check_name,
                check_display_name=result.displayName,
                severity=result.severity,
                issue=issue,
            )
update_category_option_combos() async

Trigger DHIS2 to (re)generate the CategoryOptionCombo matrix.

DHIS2 v42 auto-generated COCs whenever a CategoryCombo was saved, so callers rarely needed this. v43 changed the behavior — saving a CategoryCombo no longer triggers regeneration; the matrix stays empty until this maintenance task runs.

client.category_combos.wait_for_coc_generation calls this helper internally before polling so the helper "just works" on both versions. Call it directly when you need to ensure the COC matrix is up to date for an existing combo (e.g. after appending a category via add_category).

The endpoint is synchronous — DHIS2 walks every persisted combo, adds missing COCs, removes orphaned ones, and returns when done.

Source code in packages/dhis2w-client/src/dhis2w_client/v42/maintenance.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
async def update_category_option_combos(self) -> None:
    """Trigger DHIS2 to (re)generate the CategoryOptionCombo matrix.

    DHIS2 v42 auto-generated COCs whenever a CategoryCombo was saved,
    so callers rarely needed this. v43 changed the behavior — saving
    a CategoryCombo no longer triggers regeneration; the matrix stays
    empty until this maintenance task runs.

    `client.category_combos.wait_for_coc_generation` calls this
    helper internally before polling so the helper "just works" on
    both versions. Call it directly when you need to ensure the COC
    matrix is up to date for an existing combo (e.g. after appending
    a category via `add_category`).

    The endpoint is synchronous — DHIS2 walks every persisted combo,
    adds missing COCs, removes orphaned ones, and returns when done.
    """
    await self._client.post_raw("/api/maintenance/categoryOptionComboUpdate", {})