Maintenance

MaintenanceAccessor on Dhis2Client.maintenance — the data-integrity reader (get_integrity_report, iter_integrity_issues), the on-demand CategoryOptionCombo matrix regeneration (update_category_option_combos), and the analytics-table build trigger (run_analytics_tables). Predictor and validation runs live on their own accessors; see Validation + predictors.

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.
• Build the analytics tables after a data push (run_analytics_tables) and follow the job through client.tasks.

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)}")

Worked example — build the analytics tables

run_analytics_tables POSTs /api/resourceTables/analytics and returns the job-kickoff WebMessageResponse. last_years caps how many years the tables cover; each skip_* flag drops one build phase and is only sent when set. The envelope's task_ref() is what client.tasks takes, and notifier_endpoint() is the /api/system/tasks/{jobType}/{uid} path verbatim.

async with open_client(profile_from_env()) as client:
    envelope = await client.maintenance.run_analytics_tables(last_years=1, skip_outliers=True)
    ref = envelope.task_ref()
    if ref is None:
        raise RuntimeError(f"no job scheduled: {envelope.message}")
    print(f"feed at {envelope.notifier_endpoint()}")

    # Block until DHIS2 posts the terminal notification ...
    completion = await client.tasks.await_completion(ref, timeout=None)
    print(f"{completion.level}  {completion.message}")

... or read the feed one poll at a time with client.tasks.poll_once when the caller has its own clock; see Tasks module. The CLI equivalent is d2w maintenance refresh analytics --watch, and the MCP tool is maintenance_refresh_analytics; both call the plugin service, which wraps this accessor.

Related examples

• examples/client/analytics_tables_poll_once.py — run_analytics_tables followed with client.tasks.poll_once, one poll per tick.
• examples/client/task_await.py — run_analytics_tables blocked on with client.tasks.await_completion.
• examples/client/integrity_issues_stream.py — iter_integrity_issues + severity histogram + early-break scan.

maintenance

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

DataIntegrityCheck and DataIntegrityIssue come from dhis2w_client.generated.v43.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/v43/oas/data_integrity_check.py

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/v43/oas/data_integrity_issue.py

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/v43/oas/notification.py

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/v43/oas/_enums.py

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"
    HTML_PUSH_ANALYTICS = "HTML_PUSH_ANALYTICS"
    TRACKER_TRIGRAM_INDEX_MAINTENANCE = "TRACKER_TRIGRAM_INDEX_MAINTENANCE"
    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/v43/oas/_enums.py

class NotificationDataType(StrEnum):
    """NotificationDataType."""

    PARAMETERS = "PARAMETERS"

NotificationLevel

Bases: StrEnum

NotificationLevel.

Source code in packages/dhis2w-client/src/dhis2w_client/generated/v43/oas/_enums.py

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/v43/maintenance.py

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/v43/maintenance.py

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)

Methods:

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/v43/maintenance.py

@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/v43/maintenance.py

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/v43/maintenance.py

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 run_analytics_tables(
        self,
        *,
        last_years: int | None = None,
        skip_resource_tables: bool = False,
        skip_aggregate: bool = False,
        skip_events: bool = False,
        skip_enrollment: bool = False,
        skip_org_unit_ownership: bool = False,
        skip_outliers: bool = False,
        skip_tracked_entities: bool = False,
        skip_validation_result: bool = False,
    ) -> WebMessageResponse:
        """Kick off an analytics-table build; return the job-kickoff envelope.

        POSTs `/api/resourceTables/analytics`. DHIS2 schedules an
        `ANALYTICS_TABLE` job and answers with a job-kickoff
        `WebMessageResponse` — call `.task_ref()` on the result for
        `(job_type, task_uid)` and `.notifier_endpoint()` for the
        `relativeNotifierEndpoint` path. Feed the task ref to
        `client.tasks.poll_once` to read notifications one engine tick at a
        time, or `client.tasks.await_completion` to block until the build
        finishes.

        Each `skip_*` flag drops one build phase; `last_years` caps how many
        years of data the tables cover (omit for DHIS2's default). A flag left
        at its `False` default is not sent — DHIS2 already defaults it to
        `False`.
        """
        params: dict[str, Any] = {}
        if last_years is not None:
            params["lastYears"] = last_years
        if skip_resource_tables:
            params["skipResourceTables"] = "true"
        if skip_aggregate:
            params["skipAggregate"] = "true"
        if skip_events:
            params["skipEvents"] = "true"
        if skip_enrollment:
            params["skipEnrollment"] = "true"
        if skip_org_unit_ownership:
            params["skipOrgUnitOwnership"] = "true"
        if skip_outliers:
            params["skipOutliers"] = "true"
        if skip_tracked_entities:
            params["skipTrackedEntities"] = "true"
        if skip_validation_result:
            params["skipValidationResult"] = "true"
        return await self._client.post(
            "/api/resourceTables/analytics",
            body=None,
            params=params or None,
            model=WebMessageResponse,
        )

    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", {})

Methods:

__init__(client)

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

Source code in packages/dhis2w-client/src/dhis2w_client/v43/maintenance.py

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/v43/maintenance.py

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/v43/maintenance.py

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,
            )

run_analytics_tables(*, last_years=None, skip_resource_tables=False, skip_aggregate=False, skip_events=False, skip_enrollment=False, skip_org_unit_ownership=False, skip_outliers=False, skip_tracked_entities=False, skip_validation_result=False) async

Kick off an analytics-table build; return the job-kickoff envelope.

POSTs /api/resourceTables/analytics. DHIS2 schedules an ANALYTICS_TABLE job and answers with a job-kickoff WebMessageResponse — call .task_ref() on the result for (job_type, task_uid) and .notifier_endpoint() for the relativeNotifierEndpoint path. Feed the task ref to client.tasks.poll_once to read notifications one engine tick at a time, or client.tasks.await_completion to block until the build finishes.

Each skip_* flag drops one build phase; last_years caps how many years of data the tables cover (omit for DHIS2's default). A flag left at its False default is not sent — DHIS2 already defaults it to False.

Source code in packages/dhis2w-client/src/dhis2w_client/v43/maintenance.py

async def run_analytics_tables(
    self,
    *,
    last_years: int | None = None,
    skip_resource_tables: bool = False,
    skip_aggregate: bool = False,
    skip_events: bool = False,
    skip_enrollment: bool = False,
    skip_org_unit_ownership: bool = False,
    skip_outliers: bool = False,
    skip_tracked_entities: bool = False,
    skip_validation_result: bool = False,
) -> WebMessageResponse:
    """Kick off an analytics-table build; return the job-kickoff envelope.

    POSTs `/api/resourceTables/analytics`. DHIS2 schedules an
    `ANALYTICS_TABLE` job and answers with a job-kickoff
    `WebMessageResponse` — call `.task_ref()` on the result for
    `(job_type, task_uid)` and `.notifier_endpoint()` for the
    `relativeNotifierEndpoint` path. Feed the task ref to
    `client.tasks.poll_once` to read notifications one engine tick at a
    time, or `client.tasks.await_completion` to block until the build
    finishes.

    Each `skip_*` flag drops one build phase; `last_years` caps how many
    years of data the tables cover (omit for DHIS2's default). A flag left
    at its `False` default is not sent — DHIS2 already defaults it to
    `False`.
    """
    params: dict[str, Any] = {}
    if last_years is not None:
        params["lastYears"] = last_years
    if skip_resource_tables:
        params["skipResourceTables"] = "true"
    if skip_aggregate:
        params["skipAggregate"] = "true"
    if skip_events:
        params["skipEvents"] = "true"
    if skip_enrollment:
        params["skipEnrollment"] = "true"
    if skip_org_unit_ownership:
        params["skipOrgUnitOwnership"] = "true"
    if skip_outliers:
        params["skipOutliers"] = "true"
    if skip_tracked_entities:
        params["skipTrackedEntities"] = "true"
    if skip_validation_result:
        params["skipValidationResult"] = "true"
    return await self._client.post(
        "/api/resourceTables/analytics",
        body=None,
        params=params or None,
        model=WebMessageResponse,
    )

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/v43/maintenance.py

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", {})