Skip to content

Program indicators

Two accessors on Dhis2Client for the ProgramIndicator authoring surface:

Accessor API path Purpose
client.program_indicators /api/programIndicators Computed values over tracker event / enrollment data. CRUD + rename + expression validation.
client.program_indicator_groups /api/programIndicatorGroups Thematic groupings of program indicators. Per-item membership add/remove.

Unlike the aggregate indicators surface, DHIS2 does not expose a programIndicatorGroupSet resource — so this is a pair rather than the X / XGroup / XGroupSet triple used by data elements, indicators, organisation units, and (soon) category options.

No *Spec builder

Continues the design call from the org-unit / DE / indicator surfaces: keyword args on the accessor rather than a spec-over-model hop. The ProgramIndicator wire shape doesn't need the kind of transformation work that motivates a spec — see the Legend sets doc for the rule on when reaching for a *Spec is the right shape.

Expression shape

Program-indicator expressions reference event / enrollment data elements + tracked-entity attributes:

  • #{<program_uid>.<de_uid>} — one event's data-element value.
  • A{<tea_uid>} — the enrolled tracked entity's attribute value.
  • V{<program_variable>} — program-context variables (event_date, enrollment_date, org_unit, event_count, etc.).

Arithmetic + aggregation operators apply as for aggregate indicators. The optional filter expression is a boolean predicate that narrows which rows the main expression runs over.

Worked example

async with Dhis2Client(...) as client:
    # Pre-flight so DE / TEA / program UID typos surface as a 200 OK
    # with status FAILED instead of a 409 create rejection.
    desc = await client.program_indicators.validate_expression(
        "#{IpHINAT79UW.s46m5MS0hxu}",
    )
    assert desc.status == "OK", desc.message

    pi = await client.program_indicators.create(
        name="BCG per enrollment",
        short_name="BCG per enr",
        program_uid="IpHINAT79UW",
        expression="#{IpHINAT79UW.s46m5MS0hxu}",
        analytics_type="EVENT",
        filter_expression="A{child_age_in_months} < 12",
    )

    group = await client.program_indicator_groups.create(
        name="Immunisation program indicators",
        short_name="Immun PI",
    )
    await client.program_indicator_groups.add_members(
        group.id,
        program_indicator_uids=[pi.id],
    )

analytics_type picks the aggregation granularity: EVENT aggregates per event row; ENROLLMENT aggregates per enrolled tracked entity.

CLI

dhis2 metadata program-indicators list --program IpHINAT79UW
dhis2 metadata program-indicators validate-expression "#{IpHINAT79UW.s46m5MS0hxu}"
dhis2 metadata program-indicators create \
    --name "BCG per enrollment" --short-name "BCG per enr" \
    --program IpHINAT79UW \
    --expression "#{IpHINAT79UW.s46m5MS0hxu}" \
    --analytics-type EVENT
dhis2 metadata program-indicator-groups create --name "Immun PI" --short-name "Immun"
dhis2 metadata program-indicator-groups add-members <GROUP_UID> --program-indicator <PI_UID>

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

MCP

14 tools: metadata_program_indicator_{list,get,create,rename,validate_expression,set_legend_sets,delete}, metadata_program_indicator_group_{list,get,members,create,add_members,remove_members,delete}.

program_indicators

ProgramIndicator authoring — Dhis2Client.program_indicators.

ProgramIndicators are computed values over tracker event / enrollment data (the tracker analogue of aggregate Indicators). Each one carries a program reference, an analyticsType (EVENT or ENROLLMENT), and two DHIS2 expressions: expression (the numerator-shaped computation) plus an optional filter (a boolean predicate that narrows the event/enrollment set).

Generic CRUD stays on the generated accessor (client.resources.program_indicators); this module adds the authoring primitives the expression-based shape demands:

  • create(...) — named kwargs for the minimal required subset (name, short_name, program_uid, expression, analytics_type).
  • update(pi) — PUT with an existing typed model.
  • rename(uid, ...) — partial-update shortcut for label fields.
  • validate_expression(expr) — pre-flight wrapper around client.validation.describe_expression(..., context="program-indicator") to catch bad DE / TEA / attribute references before create.

Classes

ProgramIndicator

Bases: BaseModel

Generated model for DHIS2 ProgramIndicator.

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

API endpoint: /api/programIndicators.

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/program_indicator.py
class ProgramIndicator(BaseModel):
    """Generated model for DHIS2 `ProgramIndicator`.

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

    API endpoint: /api/programIndicators.

    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=255.")
    aggregateExportCategoryOptionCombo: str | None = Field(default=None, description="Length/value max=255.")
    aggregateExportDataElement: str | None = Field(default=None, description="Length/value max=255.")
    aggregationType: AggregationType | None = None
    analyticsPeriodBoundaries: list[Any] | None = Field(
        default=None, description="Collection of AnalyticsPeriodBoundary."
    )
    analyticsType: AnalyticsType | None = None
    attributeCombo: Reference | None = Field(default=None, description="Reference to CategoryCombo.")
    attributeValues: Any | None = Field(default=None, description="Reference to AttributeValues. Length/value max=255.")
    categoryCombo: Reference | None = Field(default=None, description="Reference to CategoryCombo.")
    categoryMappingIds: list[Any] | None = Field(
        default=None, description="Collection of String. 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.")
    decimals: int | None = Field(default=None, description="Length/value max=2147483647.")
    description: str | None = Field(default=None, description="Length/value min=1, 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.")
    displayInForm: bool | None = None
    displayName: str | None = Field(default=None, description="Read-only.")
    displayShortName: str | None = Field(default=None, description="Read-only.")
    expression: str | None = Field(default=None, description="Length/value max=2147483647.")
    favorite: bool | None = Field(default=None, description="Read-only.")
    favorites: list[Any] | None = Field(default=None, description="Collection of String. Read-only (inverse side).")
    filter: str | None = Field(default=None, description="Length/value max=2147483647.")
    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.")
    lastUpdated: datetime | None = None
    lastUpdatedBy: Reference | None = Field(default=None, description="Reference to User.")
    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.")
    name: str | None = Field(default=None, description="Unique. Length/value min=1, max=230.")
    orgUnitField: str | None = Field(default=None, description="Length/value max=2147483647.")
    program: Reference | None = Field(default=None, description="Reference to Program.")
    programIndicatorGroups: list[Any] | None = Field(
        default=None, description="Collection of ProgramIndicatorGroup. Read-only (inverse side)."
    )
    queryMods: Any | None = Field(default=None, description="Reference to QueryModifiers. Read-only (inverse side).")
    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.")
    style: Any | None = Field(default=None, description="Reference to ObjectStyle. 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).")

ProgramIndicatorsAccessor

Dhis2Client.program_indicators — CRUD + rename + expression validation.

Source code in packages/dhis2w-client/src/dhis2w_client/program_indicators.py
class ProgramIndicatorsAccessor:
    """`Dhis2Client.program_indicators` — CRUD + rename + expression validation."""

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

    async def list_all(
        self,
        *,
        program_uid: str | None = None,
        page: int = 1,
        page_size: int = 50,
    ) -> list[ProgramIndicator]:
        """Page through ProgramIndicators, optionally scoped to one program."""
        filters: list[str] | None = None
        if program_uid is not None:
            filters = [f"program.id:eq:{program_uid}"]
        return cast(
            list[ProgramIndicator],
            await self._client.resources.program_indicators.list(
                fields=_PI_FIELDS,
                filters=filters,
                page=page,
                page_size=page_size,
            ),
        )

    async def get(self, uid: str) -> ProgramIndicator:
        """Fetch one ProgramIndicator by UID."""
        return await self._client.get(
            f"/api/programIndicators/{uid}", model=ProgramIndicator, params={"fields": _PI_FIELDS}
        )

    async def create(
        self,
        *,
        name: str,
        short_name: str,
        program_uid: str,
        expression: str,
        analytics_type: AnalyticsType | str = AnalyticsType.EVENT,
        filter_expression: str | None = None,
        description: str | None = None,
        aggregation_type: str | None = None,
        decimals: int | None = None,
        legend_set_uids: list[str] | None = None,
        code: str | None = None,
        uid: str | None = None,
    ) -> ProgramIndicator:
        """Create a ProgramIndicator.

        `program_uid` is required — ProgramIndicators don't stand alone;
        they compute over one program's event / enrollment rows.
        `analytics_type=EVENT` aggregates per event row; `ENROLLMENT`
        aggregates per enrolled tracked entity. `expression` is the
        numerator-shaped DHIS2 expression (uses `#{<de_uid>}` for data
        elements, `A{<tea_uid>}` for tracked-entity attributes, etc.);
        `filter_expression` is an optional boolean predicate that
        narrows the rows the expression runs over. Call
        `validate_expression(expr)` first to catch typos before a
        failed create.
        """
        payload: dict[str, Any] = {
            "name": name,
            "shortName": short_name,
            "program": {"id": program_uid},
            "expression": expression,
            "analyticsType": analytics_type.value if isinstance(analytics_type, AnalyticsType) else analytics_type,
        }
        if filter_expression is not None:
            payload["filter"] = filter_expression
        if description:
            payload["description"] = description
        if aggregation_type:
            payload["aggregationType"] = aggregation_type
        if decimals is not None:
            payload["decimals"] = decimals
        if legend_set_uids:
            payload["legendSets"] = [{"id": uid_} for uid_ in legend_set_uids]
        if code:
            payload["code"] = code
        if uid:
            payload["id"] = uid
        envelope = await self._client.post("/api/programIndicators", payload, model=WebMessageResponse)
        created_uid = envelope.created_uid or uid
        if not created_uid:
            raise RuntimeError("program-indicator create did not return a uid")
        return await self.get(created_uid)

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

    async def rename(
        self,
        uid: str,
        *,
        name: str | None = None,
        short_name: str | None = None,
        description: str | None = None,
    ) -> ProgramIndicator:
        """Partial-update the label fields — read, mutate, 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 set_legend_sets(self, uid: str, *, legend_set_uids: list[str]) -> ProgramIndicator:
        """Replace the legend-set refs on one ProgramIndicator."""
        current = await self.get(uid)
        current.legendSets = [Reference(id=ref).model_dump(by_alias=True, exclude_none=True) for ref in legend_set_uids]
        return await self.update(current)

    async def validate_expression(self, expression: str) -> ExpressionDescription:
        """Parse-check a program-indicator expression via DHIS2's validator.

        Wraps `client.validation.describe_expression(expression,
        context="program-indicator")`. Returns the typed
        `ExpressionDescription` — `.status == "OK"` on success, `.message`
        names the failing reference when DHIS2 rejects. Cheap pre-flight
        for create flows that want early feedback on DE / TEA UID typos.
        """
        return await self._client.validation.describe_expression(expression, context="program-indicator")

    async def delete(self, uid: str) -> None:
        """Delete a ProgramIndicator — DHIS2 rejects deletes on PIs used in viz / dashboards."""
        if not uid:
            raise ValueError("delete requires a non-empty uid")
        await self._client.resources.program_indicators.delete(uid)
Functions
__init__(client)

Bind to the sharing client.

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

Page through ProgramIndicators, optionally scoped to one program.

Source code in packages/dhis2w-client/src/dhis2w_client/program_indicators.py
async def list_all(
    self,
    *,
    program_uid: str | None = None,
    page: int = 1,
    page_size: int = 50,
) -> list[ProgramIndicator]:
    """Page through ProgramIndicators, optionally scoped to one program."""
    filters: list[str] | None = None
    if program_uid is not None:
        filters = [f"program.id:eq:{program_uid}"]
    return cast(
        list[ProgramIndicator],
        await self._client.resources.program_indicators.list(
            fields=_PI_FIELDS,
            filters=filters,
            page=page,
            page_size=page_size,
        ),
    )
get(uid) async

Fetch one ProgramIndicator by UID.

Source code in packages/dhis2w-client/src/dhis2w_client/program_indicators.py
async def get(self, uid: str) -> ProgramIndicator:
    """Fetch one ProgramIndicator by UID."""
    return await self._client.get(
        f"/api/programIndicators/{uid}", model=ProgramIndicator, params={"fields": _PI_FIELDS}
    )
create(*, name, short_name, program_uid, expression, analytics_type=AnalyticsType.EVENT, filter_expression=None, description=None, aggregation_type=None, decimals=None, legend_set_uids=None, code=None, uid=None) async

Create a ProgramIndicator.

program_uid is required — ProgramIndicators don't stand alone; they compute over one program's event / enrollment rows. analytics_type=EVENT aggregates per event row; ENROLLMENT aggregates per enrolled tracked entity. expression is the numerator-shaped DHIS2 expression (uses #{<de_uid>} for data elements, A{<tea_uid>} for tracked-entity attributes, etc.); filter_expression is an optional boolean predicate that narrows the rows the expression runs over. Call validate_expression(expr) first to catch typos before a failed create.

Source code in packages/dhis2w-client/src/dhis2w_client/program_indicators.py
async def create(
    self,
    *,
    name: str,
    short_name: str,
    program_uid: str,
    expression: str,
    analytics_type: AnalyticsType | str = AnalyticsType.EVENT,
    filter_expression: str | None = None,
    description: str | None = None,
    aggregation_type: str | None = None,
    decimals: int | None = None,
    legend_set_uids: list[str] | None = None,
    code: str | None = None,
    uid: str | None = None,
) -> ProgramIndicator:
    """Create a ProgramIndicator.

    `program_uid` is required — ProgramIndicators don't stand alone;
    they compute over one program's event / enrollment rows.
    `analytics_type=EVENT` aggregates per event row; `ENROLLMENT`
    aggregates per enrolled tracked entity. `expression` is the
    numerator-shaped DHIS2 expression (uses `#{<de_uid>}` for data
    elements, `A{<tea_uid>}` for tracked-entity attributes, etc.);
    `filter_expression` is an optional boolean predicate that
    narrows the rows the expression runs over. Call
    `validate_expression(expr)` first to catch typos before a
    failed create.
    """
    payload: dict[str, Any] = {
        "name": name,
        "shortName": short_name,
        "program": {"id": program_uid},
        "expression": expression,
        "analyticsType": analytics_type.value if isinstance(analytics_type, AnalyticsType) else analytics_type,
    }
    if filter_expression is not None:
        payload["filter"] = filter_expression
    if description:
        payload["description"] = description
    if aggregation_type:
        payload["aggregationType"] = aggregation_type
    if decimals is not None:
        payload["decimals"] = decimals
    if legend_set_uids:
        payload["legendSets"] = [{"id": uid_} for uid_ in legend_set_uids]
    if code:
        payload["code"] = code
    if uid:
        payload["id"] = uid
    envelope = await self._client.post("/api/programIndicators", payload, model=WebMessageResponse)
    created_uid = envelope.created_uid or uid
    if not created_uid:
        raise RuntimeError("program-indicator create did not return a uid")
    return await self.get(created_uid)
update(pi) async

PUT an edited ProgramIndicator back. pi.id must be set.

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

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

Source code in packages/dhis2w-client/src/dhis2w_client/program_indicators.py
async def rename(
    self,
    uid: str,
    *,
    name: str | None = None,
    short_name: str | None = None,
    description: str | None = None,
) -> ProgramIndicator:
    """Partial-update the label fields — read, mutate, 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)
set_legend_sets(uid, *, legend_set_uids) async

Replace the legend-set refs on one ProgramIndicator.

Source code in packages/dhis2w-client/src/dhis2w_client/program_indicators.py
async def set_legend_sets(self, uid: str, *, legend_set_uids: list[str]) -> ProgramIndicator:
    """Replace the legend-set refs on one ProgramIndicator."""
    current = await self.get(uid)
    current.legendSets = [Reference(id=ref).model_dump(by_alias=True, exclude_none=True) for ref in legend_set_uids]
    return await self.update(current)
validate_expression(expression) async

Parse-check a program-indicator expression via DHIS2's validator.

Wraps client.validation.describe_expression(expression, context="program-indicator"). Returns the typed ExpressionDescription.status == "OK" on success, .message names the failing reference when DHIS2 rejects. Cheap pre-flight for create flows that want early feedback on DE / TEA UID typos.

Source code in packages/dhis2w-client/src/dhis2w_client/program_indicators.py
async def validate_expression(self, expression: str) -> ExpressionDescription:
    """Parse-check a program-indicator expression via DHIS2's validator.

    Wraps `client.validation.describe_expression(expression,
    context="program-indicator")`. Returns the typed
    `ExpressionDescription` — `.status == "OK"` on success, `.message`
    names the failing reference when DHIS2 rejects. Cheap pre-flight
    for create flows that want early feedback on DE / TEA UID typos.
    """
    return await self._client.validation.describe_expression(expression, context="program-indicator")
delete(uid) async

Delete a ProgramIndicator — DHIS2 rejects deletes on PIs used in viz / dashboards.

Source code in packages/dhis2w-client/src/dhis2w_client/program_indicators.py
async def delete(self, uid: str) -> None:
    """Delete a ProgramIndicator — DHIS2 rejects deletes on PIs used in viz / dashboards."""
    if not uid:
        raise ValueError("delete requires a non-empty uid")
    await self._client.resources.program_indicators.delete(uid)

program_indicator_groups

ProgramIndicatorGroup authoring — Dhis2Client.program_indicator_groups.

ProgramIndicatorGroups collect program indicators thematically (e.g. "Immunisation coverage", "HIV care continuum"). Smaller surface than the aggregate-indicator group-set triple — DHIS2 does not expose a ProgramIndicatorGroupSet resource, so this module only covers the group layer.

Classes

ProgramIndicatorGroup

Bases: BaseModel

Generated model for DHIS2 ProgramIndicatorGroup.

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

API endpoint: /api/programIndicatorGroups.

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/program_indicator_group.py
class ProgramIndicatorGroup(BaseModel):
    """Generated model for DHIS2 `ProgramIndicatorGroup`.

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

    API endpoint: /api/programIndicatorGroups.

    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=1, 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.")
    programIndicators: list[Any] | None = Field(default=None, description="Collection of ProgramIndicator.")
    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).")

ProgramIndicatorGroupsAccessor

Dhis2Client.program_indicator_groups — CRUD + membership helpers.

Source code in packages/dhis2w-client/src/dhis2w_client/program_indicator_groups.py
class ProgramIndicatorGroupsAccessor:
    """`Dhis2Client.program_indicator_groups` — CRUD + membership helpers."""

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

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

    async def get(self, uid: str) -> ProgramIndicatorGroup:
        """Fetch one group by UID with its member refs populated."""
        return await self._client.get(
            f"/api/programIndicatorGroups/{uid}", model=ProgramIndicatorGroup, params={"fields": _PIG_FIELDS}
        )

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

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

    async def update(self, group: ProgramIndicatorGroup) -> ProgramIndicatorGroup:
        """PUT an edited group 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/programIndicatorGroups/{group.id}", body=body)
        return await self.get(group.id)

    async def add_members(self, uid: str, *, program_indicator_uids: list[str]) -> ProgramIndicatorGroup:
        """Add ProgramIndicators to the group via the per-item POST shortcut."""
        for pi_uid in program_indicator_uids:
            await self._client.resources.program_indicator_groups.add_collection_item(uid, "programIndicators", pi_uid)
        return await self.get(uid)

    async def remove_members(self, uid: str, *, program_indicator_uids: list[str]) -> ProgramIndicatorGroup:
        """Drop ProgramIndicators from the group via the per-item DELETE shortcut."""
        for pi_uid in program_indicator_uids:
            await self._client.resources.program_indicator_groups.remove_collection_item(
                uid, "programIndicators", pi_uid
            )
        return await self.get(uid)

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

Bind to the sharing client.

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

Return every ProgramIndicatorGroup.

Source code in packages/dhis2w-client/src/dhis2w_client/program_indicator_groups.py
async def list_all(self) -> list[ProgramIndicatorGroup]:
    """Return every ProgramIndicatorGroup."""
    return cast(
        list[ProgramIndicatorGroup],
        await self._client.resources.program_indicator_groups.list(
            fields=_PIG_FIELDS,
            paging=False,
        ),
    )
get(uid) async

Fetch one group by UID with its member refs populated.

Source code in packages/dhis2w-client/src/dhis2w_client/program_indicator_groups.py
async def get(self, uid: str) -> ProgramIndicatorGroup:
    """Fetch one group by UID with its member refs populated."""
    return await self._client.get(
        f"/api/programIndicatorGroups/{uid}", model=ProgramIndicatorGroup, params={"fields": _PIG_FIELDS}
    )
list_members(uid, *, page=1, page_size=50) async

Page through ProgramIndicators inside one group.

Source code in packages/dhis2w-client/src/dhis2w_client/program_indicator_groups.py
async def list_members(
    self,
    uid: str,
    *,
    page: int = 1,
    page_size: int = 50,
) -> list[ProgramIndicator]:
    """Page through ProgramIndicators inside one group."""
    return cast(
        list[ProgramIndicator],
        await self._client.resources.program_indicators.list(
            fields=_MEMBER_FIELDS,
            filters=[f"programIndicatorGroups.id:eq:{uid}"],
            order=["name:asc"],
            page=page,
            page_size=page_size,
        ),
    )
create(*, name, short_name, uid=None, code=None, description=None) async

Create an empty group; add members afterwards via add_members.

Source code in packages/dhis2w-client/src/dhis2w_client/program_indicator_groups.py
async def create(
    self,
    *,
    name: str,
    short_name: str,
    uid: str | None = None,
    code: str | None = None,
    description: str | None = None,
) -> ProgramIndicatorGroup:
    """Create an empty group; add members afterwards via `add_members`."""
    payload: dict[str, Any] = {"name": name, "shortName": short_name}
    if uid:
        payload["id"] = uid
    if code:
        payload["code"] = code
    if description:
        payload["description"] = description
    envelope = await self._client.post("/api/programIndicatorGroups", payload, model=WebMessageResponse)
    created_uid = envelope.created_uid or uid
    if not created_uid:
        raise RuntimeError("program-indicator-group create did not return a uid")
    return await self.get(created_uid)
update(group) async

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

Source code in packages/dhis2w-client/src/dhis2w_client/program_indicator_groups.py
async def update(self, group: ProgramIndicatorGroup) -> ProgramIndicatorGroup:
    """PUT an edited group 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/programIndicatorGroups/{group.id}", body=body)
    return await self.get(group.id)
add_members(uid, *, program_indicator_uids) async

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

Source code in packages/dhis2w-client/src/dhis2w_client/program_indicator_groups.py
async def add_members(self, uid: str, *, program_indicator_uids: list[str]) -> ProgramIndicatorGroup:
    """Add ProgramIndicators to the group via the per-item POST shortcut."""
    for pi_uid in program_indicator_uids:
        await self._client.resources.program_indicator_groups.add_collection_item(uid, "programIndicators", pi_uid)
    return await self.get(uid)
remove_members(uid, *, program_indicator_uids) async

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

Source code in packages/dhis2w-client/src/dhis2w_client/program_indicator_groups.py
async def remove_members(self, uid: str, *, program_indicator_uids: list[str]) -> ProgramIndicatorGroup:
    """Drop ProgramIndicators from the group via the per-item DELETE shortcut."""
    for pi_uid in program_indicator_uids:
        await self._client.resources.program_indicator_groups.remove_collection_item(
            uid, "programIndicators", pi_uid
        )
    return await self.get(uid)
delete(uid) async

Delete the grouping row — member program indicators stay.

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