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