Skip to content

Tracker schema

The authoring flip side of dhis2 tracker register / enroll / add-event. DHIS2's tracker writes need a schema on the instance: the TrackedEntityType that names the kind of subject, and the TrackedEntityAttributes that describe the fields captured per enrolled TEI. Two accessors cover the leaf half of tracker-schema CRUD:

Accessor API path Purpose
client.tracked_entity_attributes /api/trackedEntityAttributes Atomic fields on a TEI (National ID, Given Name, DOB, …). CRUD + rename + common toggles (unique, generated, confidential, inherit, pattern).
client.tracked_entity_types /api/trackedEntityTypes The kind of TEI (Person, Case, Animal). CRUD + ordered attribute linkage through trackedEntityTypeAttributes[].
client.programs /api/programs Tracker container. Binds a TrackedEntityType, a set of TEAs on the enrollment form, a CategoryCombo, and the OUs that can capture. CRUD + add_attribute / remove_attribute for PTEA linkage + add_organisation_unit / remove_organisation_unit for OU scope.
client.program_stages /api/programStages Inner tracker-schema layer. Each stage owns an ordered programStageDataElements[] list (a join table with compulsory / displayInReports / allowFutureDate flags). CRUD + add_element / remove_element / reorder.

Scope

This page covers the full tracker-schema authoring chain: leaf resources (TrackedEntityAttribute + TrackedEntityType), the middle layer (Program + programTrackedEntityAttributes[]), and the inner layer (ProgramStage + programStageDataElements[]). Optional ProgramStageSection grouping (rarely used in the field) is still unauthored — reach for metadata patch if you need it.

TETA join table

Wiring a TEA onto a TET isn't a simple ref list — the link is a trackedEntityTypeAttributes[] entry that carries mandatory, searchable, and displayInList flags. The accessor's add_attribute / remove_attribute helpers round-trip the full TET, mutate the list, and PUT it back so those flags travel without a dedicated endpoint:

async with Dhis2Client(...) as client:
    national_id = await client.tracked_entity_attributes.create(
        name="National ID",
        short_name="NatID",
        unique=True,
        generated=True,
        pattern="RANDOM(#######)",
    )
    person = await client.tracked_entity_types.create(
        name="Person",
        short_name="Person",
        allow_audit_log=True,
        feature_type="NONE",
    )
    await client.tracked_entity_types.add_attribute(
        person.id,
        national_id.id,
        mandatory=True,
        searchable=True,
    )

Self-ref stripping

DHIS2's /api/trackedEntityTypes/{uid} read embeds trackedEntityTypeAttributes[].trackedEntityType = {id: <parent>} even though that field is the inverse side the importer rejects on PUT. The accessor strips it automatically before every update, mirroring the DataSet + DataSetElement workaround (BUGS tracker parity — same shape as _strip_self_ref_from_dse).

unique + generated + pattern

DHIS2 supports auto-generated attribute values for registration:

  • unique=True makes the value unique across the instance (National ID, passport number).
  • generated=True + pattern together mean DHIS2 auto-fills the value when a new TEI is registered.
  • Common patterns: "RANDOM(#######)" for a 7-digit random suffix, "#(ORGUNIT)(RANDOM)" to prefix the TEI's OU.

No *Spec builder

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

CLI

# TrackedEntityAttribute
dhis2 metadata tracked-entity-attributes create \
    --name "National ID" --short-name NatID --value-type TEXT \
    --unique --generated --pattern "RANDOM(#######)"

# TrackedEntityType + attribute linkage
dhis2 metadata tracked-entity-types create \
    --name Person --short-name Person --allow-audit-log --feature-type NONE
dhis2 metadata tracked-entity-types add-attribute <TET_UID> <TEA_UID> --mandatory --searchable

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

MCP

12 tools: metadata_tracked_entity_attribute_* (list / get / create / rename / delete), metadata_tracked_entity_type_* (list / get / create / rename / add-attribute / remove-attribute / delete).

Using them with tracker writes

The point of authoring these here is to make the tracker-write plugin usable end-to-end from CLI alone:

# 1. author the schema
dhis2 metadata tracked-entity-types create --name Person --short-name Person ...
# 2. use it
dhis2 tracker register --type <TET_UID> --ou <OU_UID> ...

See the tracker plugin for the write-side reference.

Program authoring

A Program binds everything together. Two flavours: WITH_REGISTRATION (tracker — requires a TET, enrolls individual TEIs) and WITHOUT_REGISTRATION (event program — captures anonymous events directly).

async with Dhis2Client(...) as client:
    program = await client.programs.create(
        name="Antenatal care",
        short_name="ANC",
        program_type="WITH_REGISTRATION",
        tracked_entity_type_uid=person.id,
        display_incident_date=True,
        only_enroll_once=True,
    )
    await client.programs.add_attribute(
        program.id,
        national_id.id,
        mandatory=True,
        searchable=True,
        sort_order=1,
    )
    await client.programs.add_organisation_unit(program.id, root_ou_uid)

PTEA join table + mergeMode=REPLACE quirk

programTrackedEntityAttributes[] is a nested join table (DHIS2's wire name is trackedEntityAttribute on the entry, not attribute). DHIS2 v42's PUT /api/programs/{uid} treats nested-list updates additively by default — items omitted from the payload are NOT removed. The accessor always passes ?mergeMode=REPLACE on PUT so remove_attribute behaves symmetrically.

OU scoping

add_organisation_unit / remove_organisation_unit use DHIS2's per-item shortcut (POST/DELETE /api/programs/{program}/organisationUnits/{ou}) — avoids the round-trip PUT entirely.

tracked_entity_attributes

TrackedEntityAttribute authoring — Dhis2Client.tracked_entity_attributes.

TrackedEntityAttributes are the atomic fields on a tracked entity (National ID, Given Name, Date of Birth, …). They get wired into a TrackedEntityType (via trackedEntityTypeAttributes[]) and/or into programs (via programTrackedEntityAttributes[]).

This module adds the CRUD primitives — the run / write side lives on Dhis2Client.tracker (the existing tracker write plugin).

  • create(...) — named kwargs over the minimal required subset (name, short_name, value_type, aggregation_type) plus the optional references (option_set_uid, legend_set_uids) and the common toggles (unique, generated, confidential, inherit, display_in_list_no_program, pattern).
  • update(tea) / rename(uid, ...) — standard edit pathways.
  • delete(uid) — DHIS2 rejects deletes on TEAs in use.

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

Classes

TrackedEntityAttribute

Bases: BaseModel

Generated model for DHIS2 TrackedEntityAttribute.

DHIS2 Tracked Entity Attribute - persisted metadata (generated from /api/schemas at DHIS2 v42).

API endpoint: /api/trackedEntityAttributes.

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/tracked_entity_attribute.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
class TrackedEntityAttribute(BaseModel):
    """Generated model for DHIS2 `TrackedEntityAttribute`.

    DHIS2 Tracked Entity Attribute - persisted metadata (generated from /api/schemas at DHIS2 v42).

    API endpoint: /api/trackedEntityAttributes.

    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).")
    aggregationType: AggregationType | None = None
    attributeValues: Any | None = Field(default=None, description="Reference to AttributeValues. Length/value max=255.")
    code: str | None = Field(default=None, description="Unique. Length/value max=50.")
    confidential: bool | None = None
    created: datetime | None = None
    createdBy: Reference | None = Field(default=None, description="Reference to User.")
    description: str | None = Field(default=None, description="Length/value min=2, max=2147483647.")
    dimensionItem: str | None = Field(default=None, description="Read-only.")
    dimensionItemType: DimensionItemType | None = None
    displayDescription: str | None = Field(default=None, description="Read-only.")
    displayFormName: str | None = Field(default=None, description="Read-only.")
    displayInListNoProgram: bool | None = None
    displayName: str | None = Field(default=None, description="Read-only.")
    displayOnVisitSchedule: bool | None = None
    displayShortName: str | None = Field(default=None, description="Read-only.")
    expression: str | None = Field(default=None, description="Length/value max=255.")
    favorite: bool | None = Field(default=None, description="Read-only.")
    favorites: list[Any] | None = Field(default=None, description="Collection of String. Read-only (inverse side).")
    fieldMask: str | None = Field(default=None, description="Length/value max=255.")
    formName: str | None = Field(default=None, description="Length/value max=2147483647.")
    generated: bool | None = None
    href: str | None = None
    id: str | None = Field(default=None, description="Unique. Length/value min=11, max=11.")
    inherit: bool | None = None
    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.")
    optionSet: Reference | None = Field(default=None, description="Reference to OptionSet.")
    optionSetValue: bool | None = Field(default=None, description="Read-only.")
    orgunitScope: bool | None = None
    pattern: str | None = Field(default=None, description="Length/value max=255.")
    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.")
    skipSynchronization: bool | None = None
    sortOrderInListNoProgram: int | None = Field(default=None, description="Length/value max=2147483647.")
    sortOrderInVisitSchedule: int | None = Field(default=None, description="Length/value max=2147483647.")
    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.")
    unique: bool | None = None
    user: Reference | None = Field(default=None, description="Reference to User. Read-only (inverse side).")
    valueType: ValueType | None = None

TrackedEntityAttributesAccessor

Dhis2Client.tracked_entity_attributes — CRUD over /api/trackedEntityAttributes.

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

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

    async def list_all(
        self,
        *,
        value_type: ValueType | str | None = None,
        page: int = 1,
        page_size: int = 50,
    ) -> list[TrackedEntityAttribute]:
        """Page through TrackedEntityAttributes, optionally filtered by valueType."""
        filters: list[str] | None = None
        if value_type is not None:
            value = value_type.value if isinstance(value_type, ValueType) else value_type
            filters = [f"valueType:eq:{value}"]
        return cast(
            list[TrackedEntityAttribute],
            await self._client.resources.tracked_entity_attributes.list(
                fields=_TEA_FIELDS,
                filters=filters,
                page=page,
                page_size=page_size,
            ),
        )

    async def get(self, uid: str) -> TrackedEntityAttribute:
        """Fetch one TrackedEntityAttribute with its optionSet + legendSet refs inline."""
        return await self._client.get(
            f"/api/trackedEntityAttributes/{uid}", model=TrackedEntityAttribute, params={"fields": _TEA_FIELDS}
        )

    async def create(
        self,
        *,
        name: str,
        short_name: str,
        value_type: ValueType | str = ValueType.TEXT,
        aggregation_type: AggregationType | str = AggregationType.NONE,
        option_set_uid: str | None = None,
        legend_set_uids: list[str] | None = None,
        unique: bool = False,
        generated: bool = False,
        confidential: bool = False,
        inherit: bool = False,
        display_in_list_no_program: bool = False,
        orgunit_scope: bool = False,
        pattern: str | None = None,
        field_mask: str | None = None,
        code: str | None = None,
        form_name: str | None = None,
        description: str | None = None,
        uid: str | None = None,
    ) -> TrackedEntityAttribute:
        """Create a TrackedEntityAttribute.

        `value_type` defaults to `TEXT`; switch to `NUMBER`, `DATE`,
        `PHONE_NUMBER`, etc. via the `ValueType` StrEnum. `unique=True`
        makes the value unique across the instance (National ID,
        passport number). `generated=True` + `pattern` lets DHIS2
        auto-generate the value when a TEI is registered (common for
        case IDs). `option_set_uid` constrains to an option-set
        picklist. `confidential=True` tags the attribute as sensitive
        for audit + sharing policies.
        """
        payload: dict[str, Any] = {
            "name": name,
            "shortName": short_name,
            "valueType": value_type.value if isinstance(value_type, ValueType) else value_type,
            "aggregationType": (
                aggregation_type.value if isinstance(aggregation_type, AggregationType) else aggregation_type
            ),
            "unique": unique,
            "generated": generated,
            "confidential": confidential,
            "inherit": inherit,
            "displayInListNoProgram": display_in_list_no_program,
            "orgunitScope": orgunit_scope,
        }
        if option_set_uid:
            payload["optionSet"] = {"id": option_set_uid}
        if legend_set_uids:
            payload["legendSets"] = [{"id": uid_} for uid_ in legend_set_uids]
        if uid:
            payload["id"] = uid
        if code:
            payload["code"] = code
        if form_name:
            payload["formName"] = form_name
        if description:
            payload["description"] = description
        if pattern:
            payload["pattern"] = pattern
        if field_mask:
            payload["fieldMask"] = field_mask
        envelope = await self._client.post("/api/trackedEntityAttributes", payload, model=WebMessageResponse)
        created_uid = envelope.created_uid or uid
        if not created_uid:
            raise RuntimeError("tracked-entity-attribute create did not return a uid")
        return await self.get(created_uid)

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

    async def rename(
        self,
        uid: str,
        *,
        name: str | None = None,
        short_name: str | None = None,
        form_name: str | None = None,
        description: str | None = None,
    ) -> TrackedEntityAttribute:
        """Partial-update shortcut — read, mutate the label fields, PUT."""
        if name is None and short_name is None and form_name is None and description is None:
            raise ValueError("rename requires at least one of name / short_name / form_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 form_name is not None:
            current.formName = form_name
        if description is not None:
            current.description = description
        return await self.update(current)

    async def delete(self, uid: str) -> None:
        """Delete a TrackedEntityAttribute — DHIS2 rejects deletes on TEAs wired into a TET or program."""
        if not uid:
            raise ValueError("delete requires a non-empty uid")
        await self._client.resources.tracked_entity_attributes.delete(uid)
Functions
__init__(client)

Bind to the sharing client.

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

Page through TrackedEntityAttributes, optionally filtered by valueType.

Source code in packages/dhis2w-client/src/dhis2w_client/tracked_entity_attributes.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
async def list_all(
    self,
    *,
    value_type: ValueType | str | None = None,
    page: int = 1,
    page_size: int = 50,
) -> list[TrackedEntityAttribute]:
    """Page through TrackedEntityAttributes, optionally filtered by valueType."""
    filters: list[str] | None = None
    if value_type is not None:
        value = value_type.value if isinstance(value_type, ValueType) else value_type
        filters = [f"valueType:eq:{value}"]
    return cast(
        list[TrackedEntityAttribute],
        await self._client.resources.tracked_entity_attributes.list(
            fields=_TEA_FIELDS,
            filters=filters,
            page=page,
            page_size=page_size,
        ),
    )
get(uid) async

Fetch one TrackedEntityAttribute with its optionSet + legendSet refs inline.

Source code in packages/dhis2w-client/src/dhis2w_client/tracked_entity_attributes.py 
70
71
72
73
74
async def get(self, uid: str) -> TrackedEntityAttribute:
    """Fetch one TrackedEntityAttribute with its optionSet + legendSet refs inline."""
    return await self._client.get(
        f"/api/trackedEntityAttributes/{uid}", model=TrackedEntityAttribute, params={"fields": _TEA_FIELDS}
    )
create(*, name, short_name, value_type=ValueType.TEXT, aggregation_type=AggregationType.NONE, option_set_uid=None, legend_set_uids=None, unique=False, generated=False, confidential=False, inherit=False, display_in_list_no_program=False, orgunit_scope=False, pattern=None, field_mask=None, code=None, form_name=None, description=None, uid=None) async

Create a TrackedEntityAttribute.

value_type defaults to TEXT; switch to NUMBER, DATE, PHONE_NUMBER, etc. via the ValueType StrEnum. unique=True makes the value unique across the instance (National ID, passport number). generated=True + pattern lets DHIS2 auto-generate the value when a TEI is registered (common for case IDs). option_set_uid constrains to an option-set picklist. confidential=True tags the attribute as sensitive for audit + sharing policies.

Source code in packages/dhis2w-client/src/dhis2w_client/tracked_entity_attributes.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
async def create(
    self,
    *,
    name: str,
    short_name: str,
    value_type: ValueType | str = ValueType.TEXT,
    aggregation_type: AggregationType | str = AggregationType.NONE,
    option_set_uid: str | None = None,
    legend_set_uids: list[str] | None = None,
    unique: bool = False,
    generated: bool = False,
    confidential: bool = False,
    inherit: bool = False,
    display_in_list_no_program: bool = False,
    orgunit_scope: bool = False,
    pattern: str | None = None,
    field_mask: str | None = None,
    code: str | None = None,
    form_name: str | None = None,
    description: str | None = None,
    uid: str | None = None,
) -> TrackedEntityAttribute:
    """Create a TrackedEntityAttribute.

    `value_type` defaults to `TEXT`; switch to `NUMBER`, `DATE`,
    `PHONE_NUMBER`, etc. via the `ValueType` StrEnum. `unique=True`
    makes the value unique across the instance (National ID,
    passport number). `generated=True` + `pattern` lets DHIS2
    auto-generate the value when a TEI is registered (common for
    case IDs). `option_set_uid` constrains to an option-set
    picklist. `confidential=True` tags the attribute as sensitive
    for audit + sharing policies.
    """
    payload: dict[str, Any] = {
        "name": name,
        "shortName": short_name,
        "valueType": value_type.value if isinstance(value_type, ValueType) else value_type,
        "aggregationType": (
            aggregation_type.value if isinstance(aggregation_type, AggregationType) else aggregation_type
        ),
        "unique": unique,
        "generated": generated,
        "confidential": confidential,
        "inherit": inherit,
        "displayInListNoProgram": display_in_list_no_program,
        "orgunitScope": orgunit_scope,
    }
    if option_set_uid:
        payload["optionSet"] = {"id": option_set_uid}
    if legend_set_uids:
        payload["legendSets"] = [{"id": uid_} for uid_ in legend_set_uids]
    if uid:
        payload["id"] = uid
    if code:
        payload["code"] = code
    if form_name:
        payload["formName"] = form_name
    if description:
        payload["description"] = description
    if pattern:
        payload["pattern"] = pattern
    if field_mask:
        payload["fieldMask"] = field_mask
    envelope = await self._client.post("/api/trackedEntityAttributes", payload, model=WebMessageResponse)
    created_uid = envelope.created_uid or uid
    if not created_uid:
        raise RuntimeError("tracked-entity-attribute create did not return a uid")
    return await self.get(created_uid)
update(attribute) async

PUT an edited TrackedEntityAttribute back. attribute.id must be set.

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

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

Source code in packages/dhis2w-client/src/dhis2w_client/tracked_entity_attributes.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
async def rename(
    self,
    uid: str,
    *,
    name: str | None = None,
    short_name: str | None = None,
    form_name: str | None = None,
    description: str | None = None,
) -> TrackedEntityAttribute:
    """Partial-update shortcut — read, mutate the label fields, PUT."""
    if name is None and short_name is None and form_name is None and description is None:
        raise ValueError("rename requires at least one of name / short_name / form_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 form_name is not None:
        current.formName = form_name
    if description is not None:
        current.description = description
    return await self.update(current)
delete(uid) async

Delete a TrackedEntityAttribute — DHIS2 rejects deletes on TEAs wired into a TET or program.

Source code in packages/dhis2w-client/src/dhis2w_client/tracked_entity_attributes.py 
176
177
178
179
180
async def delete(self, uid: str) -> None:
    """Delete a TrackedEntityAttribute — DHIS2 rejects deletes on TEAs wired into a TET or program."""
    if not uid:
        raise ValueError("delete requires a non-empty uid")
    await self._client.resources.tracked_entity_attributes.delete(uid)

tracked_entity_types

TrackedEntityType authoring — Dhis2Client.tracked_entity_types.

A TrackedEntityType is the kind of subject a tracker program enrols (Person, Case, Animal). Each TET carries its own set of attributes via trackedEntityTypeAttributes[] — a join table that flags which TEAs are mandatory, searchable, and visible in the enrollment UI.

  • create(...) — named kwargs over the minimal required subset (name, short_name) plus common knobs (description, allow_audit_log, feature_type, min_attributes_required_to_search).
  • add_attribute(tet_uid, tea_uid, *, mandatory=False, searchable=False, display_in_list=True) — wire a TEA onto the TET by round-tripping the full TET, mutating trackedEntityTypeAttributes[], and PUTing back. Mirrors DataSet + DataSetElement.
  • remove_attribute(tet_uid, tea_uid) — drops the TEA ref from the TET.
  • rename(uid, ...) / delete(uid) — standard edit pathways.

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

Classes

TrackedEntityType

Bases: BaseModel

Generated model for DHIS2 TrackedEntityType.

DHIS2 Tracked Entity Type - persisted metadata (generated from /api/schemas at DHIS2 v42).

API endpoint: /api/trackedEntityTypes.

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/tracked_entity_type.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
class TrackedEntityType(BaseModel):
    """Generated model for DHIS2 `TrackedEntityType`.

    DHIS2 Tracked Entity Type - persisted metadata (generated from /api/schemas at DHIS2 v42).

    API endpoint: /api/trackedEntityTypes.

    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).")
    allowAuditLog: bool | None = None
    attributeValues: Any | None = Field(default=None, description="Reference to AttributeValues. Length/value max=255.")
    code: str | None = Field(default=None, description="Unique. Length/value max=50.")
    created: datetime | None = None
    createdBy: Reference | None = Field(default=None, description="Reference to User.")
    description: str | None = Field(default=None, description="Length/value min=1, max=2147483647.")
    displayDescription: str | None = Field(default=None, description="Read-only.")
    displayFormName: str | None = Field(default=None, description="Read-only.")
    displayName: str | None = Field(default=None, description="Read-only.")
    displayShortName: str | None = Field(default=None, description="Read-only.")
    favorite: bool | None = Field(default=None, description="Read-only.")
    favorites: list[Any] | None = Field(default=None, description="Collection of String. Read-only (inverse side).")
    featureType: FeatureType | None = None
    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.")
    maxTeiCountToReturn: int | None = Field(default=None, description="Length/value max=2147483647.")
    minAttributesRequiredToSearch: int | None = Field(default=None, description="Length/value max=2147483647.")
    name: str | None = Field(default=None, description="Unique. Length/value min=1, max=230.")
    sharing: Any | None = Field(default=None, description="Reference to Sharing. Length/value max=255.")
    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.")
    trackedEntityTypeAttributes: list[Any] | None = Field(
        default=None, description="Collection of TrackedEntityTypeAttribute."
    )
    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).")

TrackedEntityTypesAccessor

Dhis2Client.tracked_entity_types — CRUD + attribute-linkage helpers.

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

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

    async def list_all(
        self,
        *,
        page: int = 1,
        page_size: int = 50,
    ) -> list[TrackedEntityType]:
        """Page through TrackedEntityTypes."""
        raw = await self._client.get_raw(
            "/api/trackedEntityTypes",
            params={
                "fields": _TET_FIELDS,
                "page": str(page),
                "pageSize": str(page_size),
            },
        )
        rows = raw.get("trackedEntityTypes") or []
        return [TrackedEntityType.model_validate(row) for row in rows if isinstance(row, dict)]

    async def get(self, uid: str) -> TrackedEntityType:
        """Fetch one TrackedEntityType with its TEA link table resolved inline."""
        return await self._client.get(
            f"/api/trackedEntityTypes/{uid}", model=TrackedEntityType, params={"fields": _TET_FIELDS}
        )

    async def create(
        self,
        *,
        name: str,
        short_name: str,
        description: str | None = None,
        code: str | None = None,
        form_name: str | None = None,
        allow_audit_log: bool | None = None,
        feature_type: str | None = None,
        min_attributes_required_to_search: int | None = None,
        max_tei_count_to_return: int | None = None,
        uid: str | None = None,
    ) -> TrackedEntityType:
        """Create a TrackedEntityType.

        `allow_audit_log` enables the per-TEI audit trail (required for
        compliance workflows). `feature_type` governs the geometry
        attached to each TEI (`NONE` / `POINT` / `POLYGON`).
        `min_attributes_required_to_search` sets the enrollment-search
        minimum attribute count; higher values reduce accidental
        full-table scans.
        """
        payload: dict[str, Any] = {"name": name, "shortName": short_name}
        if description:
            payload["description"] = description
        if code:
            payload["code"] = code
        if form_name:
            payload["formName"] = form_name
        if allow_audit_log is not None:
            payload["allowAuditLog"] = allow_audit_log
        if feature_type:
            payload["featureType"] = feature_type
        if min_attributes_required_to_search is not None:
            payload["minAttributesRequiredToSearch"] = min_attributes_required_to_search
        if max_tei_count_to_return is not None:
            payload["maxTeiCountToReturn"] = max_tei_count_to_return
        if uid:
            payload["id"] = uid
        envelope = await self._client.post("/api/trackedEntityTypes", payload, model=WebMessageResponse)
        created_uid = envelope.created_uid or uid
        if not created_uid:
            raise RuntimeError("tracked-entity-type create did not return a uid")
        return await self.get(created_uid)

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

    async def rename(
        self,
        uid: str,
        *,
        name: str | None = None,
        short_name: str | None = None,
        form_name: str | None = None,
        description: str | None = None,
    ) -> TrackedEntityType:
        """Partial-update shortcut — read, mutate the label fields, PUT."""
        if name is None and short_name is None and form_name is None and description is None:
            raise ValueError("rename requires at least one of name / short_name / form_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 form_name is not None:
            current.formName = form_name
        if description is not None:
            current.description = description
        return await self.update(current)

    async def add_attribute(
        self,
        tet_uid: str,
        attribute_uid: str,
        *,
        mandatory: bool = False,
        searchable: bool = False,
        display_in_list: bool = True,
    ) -> TrackedEntityType:
        """Wire a TrackedEntityAttribute onto the TET.

        DHIS2 stores the link in `trackedEntityTypeAttributes[]` as a
        nested join object — the accessor round-trips the full TET,
        appends a new entry, and PUTs it back.
        """
        current = await self.get(tet_uid)
        raw = current.model_dump(by_alias=True, exclude_none=True, mode="json")
        _strip_self_ref_from_teta(raw)
        existing = raw.get("trackedEntityTypeAttributes") or []
        if any(
            (entry.get("trackedEntityAttribute") or {}).get("id") == attribute_uid
            for entry in existing
            if isinstance(entry, dict)
        ):
            return current
        existing.append(
            {
                "trackedEntityAttribute": {"id": attribute_uid},
                "mandatory": mandatory,
                "searchable": searchable,
                "displayInList": display_in_list,
            },
        )
        raw["trackedEntityTypeAttributes"] = existing
        await self._client.put_raw(f"/api/trackedEntityTypes/{tet_uid}", body=raw)
        return await self.get(tet_uid)

    async def remove_attribute(self, tet_uid: str, attribute_uid: str) -> TrackedEntityType:
        """Drop a TrackedEntityAttribute from the TET's link table."""
        current = await self.get(tet_uid)
        raw = current.model_dump(by_alias=True, exclude_none=True, mode="json")
        _strip_self_ref_from_teta(raw)
        existing = raw.get("trackedEntityTypeAttributes") or []
        filtered = [
            entry
            for entry in existing
            if isinstance(entry, dict) and (entry.get("trackedEntityAttribute") or {}).get("id") != attribute_uid
        ]
        if len(filtered) == len(existing):
            return current
        raw["trackedEntityTypeAttributes"] = filtered
        await self._client.put_raw(f"/api/trackedEntityTypes/{tet_uid}", body=raw)
        return await self.get(tet_uid)

    async def delete(self, uid: str) -> None:
        """Delete a TrackedEntityType — DHIS2 rejects deletes on TETs in use by enrolled TEIs."""
        if not uid:
            raise ValueError("delete requires a non-empty uid")
        await self._client.resources.tracked_entity_types.delete(uid)
Functions
__init__(client)

Bind to the sharing client.

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

Page through TrackedEntityTypes.

Source code in packages/dhis2w-client/src/dhis2w_client/tracked_entity_types.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
async def list_all(
    self,
    *,
    page: int = 1,
    page_size: int = 50,
) -> list[TrackedEntityType]:
    """Page through TrackedEntityTypes."""
    raw = await self._client.get_raw(
        "/api/trackedEntityTypes",
        params={
            "fields": _TET_FIELDS,
            "page": str(page),
            "pageSize": str(page_size),
        },
    )
    rows = raw.get("trackedEntityTypes") or []
    return [TrackedEntityType.model_validate(row) for row in rows if isinstance(row, dict)]
get(uid) async

Fetch one TrackedEntityType with its TEA link table resolved inline.

Source code in packages/dhis2w-client/src/dhis2w_client/tracked_entity_types.py 
67
68
69
70
71
async def get(self, uid: str) -> TrackedEntityType:
    """Fetch one TrackedEntityType with its TEA link table resolved inline."""
    return await self._client.get(
        f"/api/trackedEntityTypes/{uid}", model=TrackedEntityType, params={"fields": _TET_FIELDS}
    )
create(*, name, short_name, description=None, code=None, form_name=None, allow_audit_log=None, feature_type=None, min_attributes_required_to_search=None, max_tei_count_to_return=None, uid=None) async

Create a TrackedEntityType.

allow_audit_log enables the per-TEI audit trail (required for compliance workflows). feature_type governs the geometry attached to each TEI (NONE / POINT / POLYGON). min_attributes_required_to_search sets the enrollment-search minimum attribute count; higher values reduce accidental full-table scans.

Source code in packages/dhis2w-client/src/dhis2w_client/tracked_entity_types.py 
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
async def create(
    self,
    *,
    name: str,
    short_name: str,
    description: str | None = None,
    code: str | None = None,
    form_name: str | None = None,
    allow_audit_log: bool | None = None,
    feature_type: str | None = None,
    min_attributes_required_to_search: int | None = None,
    max_tei_count_to_return: int | None = None,
    uid: str | None = None,
) -> TrackedEntityType:
    """Create a TrackedEntityType.

    `allow_audit_log` enables the per-TEI audit trail (required for
    compliance workflows). `feature_type` governs the geometry
    attached to each TEI (`NONE` / `POINT` / `POLYGON`).
    `min_attributes_required_to_search` sets the enrollment-search
    minimum attribute count; higher values reduce accidental
    full-table scans.
    """
    payload: dict[str, Any] = {"name": name, "shortName": short_name}
    if description:
        payload["description"] = description
    if code:
        payload["code"] = code
    if form_name:
        payload["formName"] = form_name
    if allow_audit_log is not None:
        payload["allowAuditLog"] = allow_audit_log
    if feature_type:
        payload["featureType"] = feature_type
    if min_attributes_required_to_search is not None:
        payload["minAttributesRequiredToSearch"] = min_attributes_required_to_search
    if max_tei_count_to_return is not None:
        payload["maxTeiCountToReturn"] = max_tei_count_to_return
    if uid:
        payload["id"] = uid
    envelope = await self._client.post("/api/trackedEntityTypes", payload, model=WebMessageResponse)
    created_uid = envelope.created_uid or uid
    if not created_uid:
        raise RuntimeError("tracked-entity-type create did not return a uid")
    return await self.get(created_uid)
update(tet) async

PUT an edited TrackedEntityType back. tet.id must be set.

Source code in packages/dhis2w-client/src/dhis2w_client/tracked_entity_types.py 
119
120
121
122
123
124
125
126
async def update(self, tet: TrackedEntityType) -> TrackedEntityType:
    """PUT an edited TrackedEntityType back. `tet.id` must be set."""
    if not tet.id:
        raise ValueError("update requires tet.id to be set")
    body = tet.model_dump(by_alias=True, exclude_none=True, mode="json")
    _strip_self_ref_from_teta(body)
    await self._client.put_raw(f"/api/trackedEntityTypes/{tet.id}", body=body)
    return await self.get(tet.id)
rename(uid, *, name=None, short_name=None, form_name=None, description=None) async

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

Source code in packages/dhis2w-client/src/dhis2w_client/tracked_entity_types.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
async def rename(
    self,
    uid: str,
    *,
    name: str | None = None,
    short_name: str | None = None,
    form_name: str | None = None,
    description: str | None = None,
) -> TrackedEntityType:
    """Partial-update shortcut — read, mutate the label fields, PUT."""
    if name is None and short_name is None and form_name is None and description is None:
        raise ValueError("rename requires at least one of name / short_name / form_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 form_name is not None:
        current.formName = form_name
    if description is not None:
        current.description = description
    return await self.update(current)
add_attribute(tet_uid, attribute_uid, *, mandatory=False, searchable=False, display_in_list=True) async

Wire a TrackedEntityAttribute onto the TET.

DHIS2 stores the link in trackedEntityTypeAttributes[] as a nested join object — the accessor round-trips the full TET, appends a new entry, and PUTs it back.

Source code in packages/dhis2w-client/src/dhis2w_client/tracked_entity_types.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
async def add_attribute(
    self,
    tet_uid: str,
    attribute_uid: str,
    *,
    mandatory: bool = False,
    searchable: bool = False,
    display_in_list: bool = True,
) -> TrackedEntityType:
    """Wire a TrackedEntityAttribute onto the TET.

    DHIS2 stores the link in `trackedEntityTypeAttributes[]` as a
    nested join object — the accessor round-trips the full TET,
    appends a new entry, and PUTs it back.
    """
    current = await self.get(tet_uid)
    raw = current.model_dump(by_alias=True, exclude_none=True, mode="json")
    _strip_self_ref_from_teta(raw)
    existing = raw.get("trackedEntityTypeAttributes") or []
    if any(
        (entry.get("trackedEntityAttribute") or {}).get("id") == attribute_uid
        for entry in existing
        if isinstance(entry, dict)
    ):
        return current
    existing.append(
        {
            "trackedEntityAttribute": {"id": attribute_uid},
            "mandatory": mandatory,
            "searchable": searchable,
            "displayInList": display_in_list,
        },
    )
    raw["trackedEntityTypeAttributes"] = existing
    await self._client.put_raw(f"/api/trackedEntityTypes/{tet_uid}", body=raw)
    return await self.get(tet_uid)
remove_attribute(tet_uid, attribute_uid) async

Drop a TrackedEntityAttribute from the TET's link table.

Source code in packages/dhis2w-client/src/dhis2w_client/tracked_entity_types.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
async def remove_attribute(self, tet_uid: str, attribute_uid: str) -> TrackedEntityType:
    """Drop a TrackedEntityAttribute from the TET's link table."""
    current = await self.get(tet_uid)
    raw = current.model_dump(by_alias=True, exclude_none=True, mode="json")
    _strip_self_ref_from_teta(raw)
    existing = raw.get("trackedEntityTypeAttributes") or []
    filtered = [
        entry
        for entry in existing
        if isinstance(entry, dict) and (entry.get("trackedEntityAttribute") or {}).get("id") != attribute_uid
    ]
    if len(filtered) == len(existing):
        return current
    raw["trackedEntityTypeAttributes"] = filtered
    await self._client.put_raw(f"/api/trackedEntityTypes/{tet_uid}", body=raw)
    return await self.get(tet_uid)
delete(uid) async

Delete a TrackedEntityType — DHIS2 rejects deletes on TETs in use by enrolled TEIs.

Source code in packages/dhis2w-client/src/dhis2w_client/tracked_entity_types.py 
205
206
207
208
209
async def delete(self, uid: str) -> None:
    """Delete a TrackedEntityType — DHIS2 rejects deletes on TETs in use by enrolled TEIs."""
    if not uid:
        raise ValueError("delete requires a non-empty uid")
    await self._client.resources.tracked_entity_types.delete(uid)

programs

Program authoring — Dhis2Client.programs.

A DHIS2 Program is the tracker container: it binds a TrackedEntityType (for WITH_REGISTRATION programs), a set of TrackedEntityAttributes shown on the enrollment form, and the OrganisationUnits that can capture enrollments or events. Programs come in two flavours:

  • WITH_REGISTRATION — tracker programs. Need a trackedEntityType and register individual TEIs before capturing events.
  • WITHOUT_REGISTRATION — event programs. Capture anonymous events directly; no TET required.

This module is the authoring flip side of the existing tracker-write plugin (dhis2 tracker register / enroll / add-event). The leaf half (TET + TEA) lives in tracked_entity_types.py / tracked_entity_attributes.py; the inner layer (ProgramStage + PSDE) ships in a follow-up.

Surface: - create(...) — named kwargs over the minimal required subset (name, short_name, program_type) plus the refs + common knobs. For WITH_REGISTRATION, tracked_entity_type_uid is required. - add_attribute(program_uid, tea_uid, ...) — wire a TEA into the enrollment form via the programTrackedEntityAttributes[] join table. - add_organisation_unit(program_uid, ou_uid) — scope the program to an OU. Tracker writes need at least one OU in scope to register. - rename(uid, ...) / delete(uid) — standard pathways.

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

Classes

Program

Bases: BaseModel

Generated model for DHIS2 Program.

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

API endpoint: /api/programs.

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.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
class Program(BaseModel):
    """Generated model for DHIS2 `Program`.

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

    API endpoint: /api/programs.

    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).")
    accessLevel: AccessLevel | None = None
    attributeValues: Any | None = Field(default=None, description="Reference to AttributeValues. Length/value max=255.")
    categoryCombo: Reference | None = Field(default=None, description="Reference to CategoryCombo.")
    categoryMappings: list[Any] | None = Field(
        default=None, description="Collection of ProgramCategoryMapping. Length/value max=255."
    )
    code: str | None = Field(default=None, description="Unique. Length/value max=50.")
    completeEventsExpiryDays: int | None = Field(default=None, description="Length/value max=2147483647.")
    created: datetime | None = None
    createdBy: Reference | None = Field(default=None, description="Reference to User.")
    dataEntryForm: Reference | None = Field(default=None, description="Reference to DataEntryForm.")
    description: str | None = Field(default=None, description="Length/value min=1, max=2147483647.")
    displayDescription: str | None = Field(default=None, description="Read-only.")
    displayEnrollmentDateLabel: str | None = Field(default=None, description="Read-only.")
    displayEnrollmentLabel: str | None = Field(default=None, description="Read-only.")
    displayEventLabel: str | None = Field(default=None, description="Read-only.")
    displayFollowUpLabel: str | None = Field(default=None, description="Read-only.")
    displayFormName: str | None = Field(default=None, description="Read-only.")
    displayFrontPageList: bool | None = None
    displayIncidentDate: bool | None = None
    displayIncidentDateLabel: str | None = Field(default=None, description="Read-only.")
    displayName: str | None = Field(default=None, description="Read-only.")
    displayNoteLabel: str | None = Field(default=None, description="Read-only.")
    displayOrgUnitLabel: str | None = Field(default=None, description="Read-only.")
    displayProgramStageLabel: str | None = Field(default=None, description="Read-only.")
    displayRelationshipLabel: str | None = Field(default=None, description="Read-only.")
    displayShortName: str | None = Field(default=None, description="Read-only.")
    displayTrackedEntityAttributeLabel: str | None = Field(default=None, description="Read-only.")
    enrollmentDateLabel: str | None = Field(default=None, description="Length/value min=2, max=2147483647.")
    enrollmentLabel: str | None = Field(default=None, description="Length/value min=2, max=2147483647.")
    eventLabel: str | None = Field(default=None, description="Length/value min=2, max=2147483647.")
    expiryDays: int | None = Field(default=None, description="Length/value max=2147483647.")
    expiryPeriodType: PeriodType | None = Field(
        default=None, description="Reference to PeriodType. Length/value max=255."
    )
    favorite: bool | None = Field(default=None, description="Read-only.")
    favorites: list[Any] | None = Field(default=None, description="Collection of String. Read-only (inverse side).")
    featureType: FeatureType | None = None
    followUpLabel: str | None = Field(default=None, description="Length/value min=2, 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.")
    ignoreOverdueEvents: bool | None = None
    incidentDateLabel: str | None = Field(default=None, description="Length/value min=2, max=2147483647.")
    lastUpdated: datetime | None = None
    lastUpdatedBy: Reference | None = Field(default=None, description="Reference to User.")
    maxTeiCountToReturn: int | None = Field(default=None, description="Length/value max=2147483647.")
    minAttributesRequiredToSearch: int | None = Field(default=None, description="Length/value max=2147483647.")
    name: str | None = Field(default=None, description="Length/value min=1, max=230.")
    noteLabel: str | None = Field(default=None, description="Length/value min=2, max=2147483647.")
    notificationTemplates: list[Any] | None = Field(
        default=None, description="Collection of ProgramNotificationTemplate."
    )
    onlyEnrollOnce: bool | None = None
    openDaysAfterCoEndDate: int | None = Field(default=None, description="Length/value max=2147483647.")
    orgUnitLabel: str | None = Field(default=None, description="Length/value min=2, max=2147483647.")
    organisationUnits: list[Any] | None = Field(default=None, description="Collection of OrganisationUnit.")
    programIndicators: list[Any] | None = Field(
        default=None, description="Collection of ProgramIndicator. Read-only (inverse side)."
    )
    programRuleVariables: list[Any] | None = Field(
        default=None, description="Collection of ProgramRuleVariable. Read-only (inverse side)."
    )
    programSections: list[Any] | None = Field(default=None, description="Collection of ProgramSection.")
    programStageLabel: str | None = Field(default=None, description="Length/value min=2, max=2147483647.")
    programStages: list[Any] | None = Field(default=None, description="Collection of ProgramStage.")
    programTrackedEntityAttributes: list[Any] | None = Field(
        default=None, description="Collection of ProgramTrackedEntityAttribute."
    )
    programType: ProgramType | None = None
    registration: bool | None = Field(default=None, description="Read-only.")
    relatedProgram: Reference | None = Field(default=None, description="Reference to Program.")
    relationshipLabel: str | None = Field(default=None, description="Length/value min=2, max=2147483647.")
    selectEnrollmentDatesInFuture: bool | None = None
    selectIncidentDatesInFuture: bool | None = None
    sharing: Any | None = Field(default=None, description="Reference to Sharing. Length/value max=255.")
    shortName: str | None = Field(default=None, description="Length/value min=1, max=50.")
    skipOffline: bool | None = None
    style: Any | None = Field(default=None, description="Reference to ObjectStyle. Length/value max=255.")
    trackedEntityAttributeLabel: str | None = Field(default=None, description="Length/value min=2, max=2147483647.")
    trackedEntityType: Reference | None = Field(default=None, description="Reference to TrackedEntityType.")
    translations: list[Any] | None = Field(default=None, description="Collection of Translation. Length/value max=255.")
    useFirstStageDuringRegistration: bool | None = None
    user: Reference | None = Field(default=None, description="Reference to User. Read-only (inverse side).")
    userRoles: list[Any] | None = Field(default=None, description="Collection of UserRole. Read-only (inverse side).")
    version: int | None = Field(default=None, description="Length/value max=2147483647.")
    withoutRegistration: bool | None = Field(default=None, description="Read-only.")

ProgramsAccessor

Dhis2Client.programs — CRUD + attribute + OU linkage over /api/programs.

Source code in packages/dhis2w-client/src/dhis2w_client/programs.py 
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
class ProgramsAccessor:
    """`Dhis2Client.programs` — CRUD + attribute + OU linkage over `/api/programs`."""

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

    async def list_all(
        self,
        *,
        program_type: ProgramType | str | None = None,
        page: int = 1,
        page_size: int = 50,
    ) -> list[Program]:
        """Page through Programs, optionally filtered by programType."""
        filters: list[str] | None = None
        if program_type is not None:
            value = program_type.value if isinstance(program_type, ProgramType) else program_type
            filters = [f"programType:eq:{value}"]
        return cast(
            list[Program],
            await self._client.resources.programs.list(
                fields=_PROGRAM_FIELDS,
                filters=filters,
                page=page,
                page_size=page_size,
            ),
        )

    async def get(self, uid: str) -> Program:
        """Fetch one Program with its PTEAs, OUs, and ProgramStage refs inline."""
        return await self._client.get(f"/api/programs/{uid}", model=Program, params={"fields": _PROGRAM_FIELDS})

    async def create(
        self,
        *,
        name: str,
        short_name: str,
        program_type: ProgramType | str = ProgramType.WITH_REGISTRATION,
        tracked_entity_type_uid: str | None = None,
        category_combo_uid: str | None = None,
        description: str | None = None,
        code: str | None = None,
        form_name: str | None = None,
        display_incident_date: bool | None = None,
        enrollment_date_label: str | None = None,
        incident_date_label: str | None = None,
        feature_type: str | None = None,
        only_enroll_once: bool | None = None,
        select_enrollment_dates_in_future: bool | None = None,
        select_incident_dates_in_future: bool | None = None,
        expiry_days: int | None = None,
        min_attributes_required_to_search: int | None = None,
        max_tei_count_to_return: int | None = None,
        use_first_stage_during_registration: bool | None = None,
        uid: str | None = None,
    ) -> Program:
        """Create a Program.

        `program_type=WITH_REGISTRATION` (default) requires
        `tracked_entity_type_uid`; pass `WITHOUT_REGISTRATION` for an
        event program that skips TEI registration.
        `category_combo_uid` defaults to the instance-wide default
        combo (DHIS2 rejects programs without a CC ref).
        `display_incident_date` + `incident_date_label` govern whether
        the enrollment form captures an incident date distinct from
        the enrollment date (required by some case-management flows).
        """
        resolved_type = program_type.value if isinstance(program_type, ProgramType) else program_type
        if resolved_type == "WITH_REGISTRATION" and not tracked_entity_type_uid:
            raise ValueError("WITH_REGISTRATION programs require tracked_entity_type_uid")
        default_combo = category_combo_uid or await self._client.system.default_category_combo_uid()
        payload: dict[str, Any] = {
            "name": name,
            "shortName": short_name,
            "programType": resolved_type,
            "categoryCombo": {"id": default_combo},
        }
        if tracked_entity_type_uid:
            payload["trackedEntityType"] = {"id": tracked_entity_type_uid}
        if uid:
            payload["id"] = uid
        if code:
            payload["code"] = code
        if form_name:
            payload["formName"] = form_name
        if description:
            payload["description"] = description
        if display_incident_date is not None:
            payload["displayIncidentDate"] = display_incident_date
        if enrollment_date_label:
            payload["enrollmentDateLabel"] = enrollment_date_label
        if incident_date_label:
            payload["incidentDateLabel"] = incident_date_label
        if feature_type:
            payload["featureType"] = feature_type
        if only_enroll_once is not None:
            payload["onlyEnrollOnce"] = only_enroll_once
        if select_enrollment_dates_in_future is not None:
            payload["selectEnrollmentDatesInFuture"] = select_enrollment_dates_in_future
        if select_incident_dates_in_future is not None:
            payload["selectIncidentDatesInFuture"] = select_incident_dates_in_future
        if expiry_days is not None:
            payload["expiryDays"] = expiry_days
        if min_attributes_required_to_search is not None:
            payload["minAttributesRequiredToSearch"] = min_attributes_required_to_search
        if max_tei_count_to_return is not None:
            payload["maxTeiCountToReturn"] = max_tei_count_to_return
        if use_first_stage_during_registration is not None:
            payload["useFirstStageDuringRegistration"] = use_first_stage_during_registration
        envelope = await self._client.post("/api/programs", payload, model=WebMessageResponse)
        created_uid = envelope.created_uid or uid
        if not created_uid:
            raise RuntimeError("program create did not return a uid")
        return await self.get(created_uid)

    async def update(self, program: Program) -> Program:
        """PUT an edited Program back. `program.id` must be set.

        DHIS2 v42's Program PUT importer treats nested-list updates
        additively without `mergeMode=REPLACE` — items omitted from a
        list aren't removed. The accessor always passes the flag so
        `add_attribute` / `remove_attribute` behave symmetrically.
        """
        if not program.id:
            raise ValueError("update requires program.id to be set")
        body = program.model_dump(by_alias=True, exclude_none=True, mode="json")
        _strip_self_ref_from_ptea(body)
        await self._client.put_raw(f"/api/programs/{program.id}", body=body, params={"mergeMode": "REPLACE"})
        return await self.get(program.id)

    async def rename(
        self,
        uid: str,
        *,
        name: str | None = None,
        short_name: str | None = None,
        form_name: str | None = None,
        description: str | None = None,
    ) -> Program:
        """Partial-update shortcut — read, mutate the label fields, PUT."""
        if name is None and short_name is None and form_name is None and description is None:
            raise ValueError("rename requires at least one of name / short_name / form_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 form_name is not None:
            current.formName = form_name
        if description is not None:
            current.description = description
        return await self.update(current)

    async def add_attribute(
        self,
        program_uid: str,
        attribute_uid: str,
        *,
        mandatory: bool = False,
        searchable: bool = False,
        display_in_list: bool = True,
        sort_order: int | None = None,
        allow_future_date: bool = False,
        render_options_as_radio: bool = False,
    ) -> Program:
        """Wire a TrackedEntityAttribute into the Program's enrollment form.

        The PTEA join table carries the enrollment-form flags
        (`mandatory`, `searchable`, `displayInList`, `sortOrder`,
        `allowFutureDate`, `renderOptionsAsRadio`). Idempotent when
        the TEA is already linked — the existing PTEA is left alone.
        """
        current = await self.get(program_uid)
        raw = current.model_dump(by_alias=True, exclude_none=True, mode="json")
        _strip_self_ref_from_ptea(raw)
        existing = raw.get("programTrackedEntityAttributes") or []
        if any(
            (entry.get("trackedEntityAttribute") or {}).get("id") == attribute_uid
            for entry in existing
            if isinstance(entry, dict)
        ):
            return current
        new_entry: dict[str, Any] = {
            "trackedEntityAttribute": {"id": attribute_uid},
            "mandatory": mandatory,
            "searchable": searchable,
            "displayInList": display_in_list,
            "allowFutureDate": allow_future_date,
            "renderOptionsAsRadio": render_options_as_radio,
        }
        if sort_order is not None:
            new_entry["sortOrder"] = sort_order
        existing.append(new_entry)
        raw["programTrackedEntityAttributes"] = existing
        await self._client.put_raw(f"/api/programs/{program_uid}", body=raw, params={"mergeMode": "REPLACE"})
        return await self.get(program_uid)

    async def remove_attribute(self, program_uid: str, attribute_uid: str) -> Program:
        """Drop a TrackedEntityAttribute from the Program's enrollment form."""
        current = await self.get(program_uid)
        raw = current.model_dump(by_alias=True, exclude_none=True, mode="json")
        _strip_self_ref_from_ptea(raw)
        existing = raw.get("programTrackedEntityAttributes") or []
        filtered = [
            entry
            for entry in existing
            if isinstance(entry, dict) and (entry.get("trackedEntityAttribute") or {}).get("id") != attribute_uid
        ]
        if len(filtered) == len(existing):
            return current
        raw["programTrackedEntityAttributes"] = filtered
        await self._client.put_raw(f"/api/programs/{program_uid}", body=raw, params={"mergeMode": "REPLACE"})
        return await self.get(program_uid)

    async def add_organisation_unit(self, program_uid: str, organisation_unit_uid: str) -> Program:
        """Scope the Program to an additional OrganisationUnit via the per-item POST shortcut."""
        await self._client.resources.programs.add_collection_item(
            program_uid,
            "organisationUnits",
            organisation_unit_uid,
        )
        return await self.get(program_uid)

    async def remove_organisation_unit(self, program_uid: str, organisation_unit_uid: str) -> Program:
        """Drop an OrganisationUnit from the Program scope via the per-item DELETE shortcut."""
        await self._client.resources.programs.remove_collection_item(
            program_uid,
            "organisationUnits",
            organisation_unit_uid,
        )
        return await self.get(program_uid)

    async def delete(self, uid: str) -> None:
        """Delete a Program — DHIS2 rejects deletes on programs with enrolled TEIs or saved events."""
        if not uid:
            raise ValueError("delete requires a non-empty uid")
        await self._client.resources.programs.delete(uid)
Functions
__init__(client)

Bind to the sharing client.

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

Page through Programs, optionally filtered by programType.

Source code in packages/dhis2w-client/src/dhis2w_client/programs.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
async def list_all(
    self,
    *,
    program_type: ProgramType | str | None = None,
    page: int = 1,
    page_size: int = 50,
) -> list[Program]:
    """Page through Programs, optionally filtered by programType."""
    filters: list[str] | None = None
    if program_type is not None:
        value = program_type.value if isinstance(program_type, ProgramType) else program_type
        filters = [f"programType:eq:{value}"]
    return cast(
        list[Program],
        await self._client.resources.programs.list(
            fields=_PROGRAM_FIELDS,
            filters=filters,
            page=page,
            page_size=page_size,
        ),
    )
get(uid) async

Fetch one Program with its PTEAs, OUs, and ProgramStage refs inline.

Source code in packages/dhis2w-client/src/dhis2w_client/programs.py 
93
94
95
async def get(self, uid: str) -> Program:
    """Fetch one Program with its PTEAs, OUs, and ProgramStage refs inline."""
    return await self._client.get(f"/api/programs/{uid}", model=Program, params={"fields": _PROGRAM_FIELDS})
create(*, name, short_name, program_type=ProgramType.WITH_REGISTRATION, tracked_entity_type_uid=None, category_combo_uid=None, description=None, code=None, form_name=None, display_incident_date=None, enrollment_date_label=None, incident_date_label=None, feature_type=None, only_enroll_once=None, select_enrollment_dates_in_future=None, select_incident_dates_in_future=None, expiry_days=None, min_attributes_required_to_search=None, max_tei_count_to_return=None, use_first_stage_during_registration=None, uid=None) async

Create a Program.

program_type=WITH_REGISTRATION (default) requires tracked_entity_type_uid; pass WITHOUT_REGISTRATION for an event program that skips TEI registration. category_combo_uid defaults to the instance-wide default combo (DHIS2 rejects programs without a CC ref). display_incident_date + incident_date_label govern whether the enrollment form captures an incident date distinct from the enrollment date (required by some case-management flows).

Source code in packages/dhis2w-client/src/dhis2w_client/programs.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
async def create(
    self,
    *,
    name: str,
    short_name: str,
    program_type: ProgramType | str = ProgramType.WITH_REGISTRATION,
    tracked_entity_type_uid: str | None = None,
    category_combo_uid: str | None = None,
    description: str | None = None,
    code: str | None = None,
    form_name: str | None = None,
    display_incident_date: bool | None = None,
    enrollment_date_label: str | None = None,
    incident_date_label: str | None = None,
    feature_type: str | None = None,
    only_enroll_once: bool | None = None,
    select_enrollment_dates_in_future: bool | None = None,
    select_incident_dates_in_future: bool | None = None,
    expiry_days: int | None = None,
    min_attributes_required_to_search: int | None = None,
    max_tei_count_to_return: int | None = None,
    use_first_stage_during_registration: bool | None = None,
    uid: str | None = None,
) -> Program:
    """Create a Program.

    `program_type=WITH_REGISTRATION` (default) requires
    `tracked_entity_type_uid`; pass `WITHOUT_REGISTRATION` for an
    event program that skips TEI registration.
    `category_combo_uid` defaults to the instance-wide default
    combo (DHIS2 rejects programs without a CC ref).
    `display_incident_date` + `incident_date_label` govern whether
    the enrollment form captures an incident date distinct from
    the enrollment date (required by some case-management flows).
    """
    resolved_type = program_type.value if isinstance(program_type, ProgramType) else program_type
    if resolved_type == "WITH_REGISTRATION" and not tracked_entity_type_uid:
        raise ValueError("WITH_REGISTRATION programs require tracked_entity_type_uid")
    default_combo = category_combo_uid or await self._client.system.default_category_combo_uid()
    payload: dict[str, Any] = {
        "name": name,
        "shortName": short_name,
        "programType": resolved_type,
        "categoryCombo": {"id": default_combo},
    }
    if tracked_entity_type_uid:
        payload["trackedEntityType"] = {"id": tracked_entity_type_uid}
    if uid:
        payload["id"] = uid
    if code:
        payload["code"] = code
    if form_name:
        payload["formName"] = form_name
    if description:
        payload["description"] = description
    if display_incident_date is not None:
        payload["displayIncidentDate"] = display_incident_date
    if enrollment_date_label:
        payload["enrollmentDateLabel"] = enrollment_date_label
    if incident_date_label:
        payload["incidentDateLabel"] = incident_date_label
    if feature_type:
        payload["featureType"] = feature_type
    if only_enroll_once is not None:
        payload["onlyEnrollOnce"] = only_enroll_once
    if select_enrollment_dates_in_future is not None:
        payload["selectEnrollmentDatesInFuture"] = select_enrollment_dates_in_future
    if select_incident_dates_in_future is not None:
        payload["selectIncidentDatesInFuture"] = select_incident_dates_in_future
    if expiry_days is not None:
        payload["expiryDays"] = expiry_days
    if min_attributes_required_to_search is not None:
        payload["minAttributesRequiredToSearch"] = min_attributes_required_to_search
    if max_tei_count_to_return is not None:
        payload["maxTeiCountToReturn"] = max_tei_count_to_return
    if use_first_stage_during_registration is not None:
        payload["useFirstStageDuringRegistration"] = use_first_stage_during_registration
    envelope = await self._client.post("/api/programs", payload, model=WebMessageResponse)
    created_uid = envelope.created_uid or uid
    if not created_uid:
        raise RuntimeError("program create did not return a uid")
    return await self.get(created_uid)
update(program) async

PUT an edited Program back. program.id must be set.

DHIS2 v42's Program PUT importer treats nested-list updates additively without mergeMode=REPLACE — items omitted from a list aren't removed. The accessor always passes the flag so add_attribute / remove_attribute behave symmetrically.

Source code in packages/dhis2w-client/src/dhis2w_client/programs.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
async def update(self, program: Program) -> Program:
    """PUT an edited Program back. `program.id` must be set.

    DHIS2 v42's Program PUT importer treats nested-list updates
    additively without `mergeMode=REPLACE` — items omitted from a
    list aren't removed. The accessor always passes the flag so
    `add_attribute` / `remove_attribute` behave symmetrically.
    """
    if not program.id:
        raise ValueError("update requires program.id to be set")
    body = program.model_dump(by_alias=True, exclude_none=True, mode="json")
    _strip_self_ref_from_ptea(body)
    await self._client.put_raw(f"/api/programs/{program.id}", body=body, params={"mergeMode": "REPLACE"})
    return await self.get(program.id)
rename(uid, *, name=None, short_name=None, form_name=None, description=None) async

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

Source code in packages/dhis2w-client/src/dhis2w_client/programs.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
async def rename(
    self,
    uid: str,
    *,
    name: str | None = None,
    short_name: str | None = None,
    form_name: str | None = None,
    description: str | None = None,
) -> Program:
    """Partial-update shortcut — read, mutate the label fields, PUT."""
    if name is None and short_name is None and form_name is None and description is None:
        raise ValueError("rename requires at least one of name / short_name / form_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 form_name is not None:
        current.formName = form_name
    if description is not None:
        current.description = description
    return await self.update(current)
add_attribute(program_uid, attribute_uid, *, mandatory=False, searchable=False, display_in_list=True, sort_order=None, allow_future_date=False, render_options_as_radio=False) async

Wire a TrackedEntityAttribute into the Program's enrollment form.

The PTEA join table carries the enrollment-form flags (mandatory, searchable, displayInList, sortOrder, allowFutureDate, renderOptionsAsRadio). Idempotent when the TEA is already linked — the existing PTEA is left alone.

Source code in packages/dhis2w-client/src/dhis2w_client/programs.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
async def add_attribute(
    self,
    program_uid: str,
    attribute_uid: str,
    *,
    mandatory: bool = False,
    searchable: bool = False,
    display_in_list: bool = True,
    sort_order: int | None = None,
    allow_future_date: bool = False,
    render_options_as_radio: bool = False,
) -> Program:
    """Wire a TrackedEntityAttribute into the Program's enrollment form.

    The PTEA join table carries the enrollment-form flags
    (`mandatory`, `searchable`, `displayInList`, `sortOrder`,
    `allowFutureDate`, `renderOptionsAsRadio`). Idempotent when
    the TEA is already linked — the existing PTEA is left alone.
    """
    current = await self.get(program_uid)
    raw = current.model_dump(by_alias=True, exclude_none=True, mode="json")
    _strip_self_ref_from_ptea(raw)
    existing = raw.get("programTrackedEntityAttributes") or []
    if any(
        (entry.get("trackedEntityAttribute") or {}).get("id") == attribute_uid
        for entry in existing
        if isinstance(entry, dict)
    ):
        return current
    new_entry: dict[str, Any] = {
        "trackedEntityAttribute": {"id": attribute_uid},
        "mandatory": mandatory,
        "searchable": searchable,
        "displayInList": display_in_list,
        "allowFutureDate": allow_future_date,
        "renderOptionsAsRadio": render_options_as_radio,
    }
    if sort_order is not None:
        new_entry["sortOrder"] = sort_order
    existing.append(new_entry)
    raw["programTrackedEntityAttributes"] = existing
    await self._client.put_raw(f"/api/programs/{program_uid}", body=raw, params={"mergeMode": "REPLACE"})
    return await self.get(program_uid)
remove_attribute(program_uid, attribute_uid) async

Drop a TrackedEntityAttribute from the Program's enrollment form.

Source code in packages/dhis2w-client/src/dhis2w_client/programs.py 
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
async def remove_attribute(self, program_uid: str, attribute_uid: str) -> Program:
    """Drop a TrackedEntityAttribute from the Program's enrollment form."""
    current = await self.get(program_uid)
    raw = current.model_dump(by_alias=True, exclude_none=True, mode="json")
    _strip_self_ref_from_ptea(raw)
    existing = raw.get("programTrackedEntityAttributes") or []
    filtered = [
        entry
        for entry in existing
        if isinstance(entry, dict) and (entry.get("trackedEntityAttribute") or {}).get("id") != attribute_uid
    ]
    if len(filtered) == len(existing):
        return current
    raw["programTrackedEntityAttributes"] = filtered
    await self._client.put_raw(f"/api/programs/{program_uid}", body=raw, params={"mergeMode": "REPLACE"})
    return await self.get(program_uid)
add_organisation_unit(program_uid, organisation_unit_uid) async

Scope the Program to an additional OrganisationUnit via the per-item POST shortcut.

Source code in packages/dhis2w-client/src/dhis2w_client/programs.py 
279
280
281
282
283
284
285
286
async def add_organisation_unit(self, program_uid: str, organisation_unit_uid: str) -> Program:
    """Scope the Program to an additional OrganisationUnit via the per-item POST shortcut."""
    await self._client.resources.programs.add_collection_item(
        program_uid,
        "organisationUnits",
        organisation_unit_uid,
    )
    return await self.get(program_uid)
remove_organisation_unit(program_uid, organisation_unit_uid) async

Drop an OrganisationUnit from the Program scope via the per-item DELETE shortcut.

Source code in packages/dhis2w-client/src/dhis2w_client/programs.py 
288
289
290
291
292
293
294
295
async def remove_organisation_unit(self, program_uid: str, organisation_unit_uid: str) -> Program:
    """Drop an OrganisationUnit from the Program scope via the per-item DELETE shortcut."""
    await self._client.resources.programs.remove_collection_item(
        program_uid,
        "organisationUnits",
        organisation_unit_uid,
    )
    return await self.get(program_uid)
delete(uid) async

Delete a Program — DHIS2 rejects deletes on programs with enrolled TEIs or saved events.

Source code in packages/dhis2w-client/src/dhis2w_client/programs.py 
297
298
299
300
301
async def delete(self, uid: str) -> None:
    """Delete a Program — DHIS2 rejects deletes on programs with enrolled TEIs or saved events."""
    if not uid:
        raise ValueError("delete requires a non-empty uid")
    await self._client.resources.programs.delete(uid)

ProgramStage authoring

Each Program owns a stage sequence (ANC 1st visit, ANC 2nd visit, …). Each stage owns an ordered programStageDataElements[] list — a join table with compulsory, displayInReports, allowFutureDate, allowProvidedElsewhere, renderOptionsAsRadio, sortOrder per entry.

stage = await client.program_stages.create(
    name="ANC 1st visit",
    program_uid=program.id,
    sort_order=1,
    repeatable=False,
    min_days_from_start=0,
    standard_interval=30,
)
await client.program_stages.add_element(
    stage.id,
    weight_de.id,
    compulsory=True,
    sort_order=0,
)
await client.program_stages.reorder(stage.id, [second_de.id, weight_de.id])

PSDE ordering helpers

  • add_element(stage_uid, de_uid, compulsory=..., sort_order=...) — appends a new PSDE entry with typed flags.
  • remove_element(stage_uid, de_uid) — drops the PSDE entry; other flags on the remaining entries are preserved.
  • reorder(stage_uid, [de_uids]) — replaces the ordered list; PSDE flags are preserved for DEs that stay in the list and sortOrder is rewritten to match the new position.

mergeMode=REPLACE quirk

DHIS2 v42's PUT /api/programStages/{uid} treats nested-list updates additively by default (same quirk as Programs). The accessor always passes ?mergeMode=REPLACE on PUT so remove_element actually removes the PSDE entry instead of silently retaining it.

PSDE self-ref strip

The generated PSDE entry carries programStage = {id: <parent>} on reads, which DHIS2's importer rejects on PUT (inverse side). Stripped automatically before every update — mirrors DataSet+DSE, TET+TETA, Program+PTEA.

program_stages

ProgramStage authoring — Dhis2Client.program_stages.

The inner layer of tracker schema authoring. A ProgramStage is a stage/visit inside a Program — ANC 1st visit, ANC 2nd visit, vaccination schedule entry, etc. Each stage owns an ordered list of programStageDataElements[] (a join table: which DEs the stage captures, in what order, with per-DE compulsory / displayInReports / allowFutureDate / allowProvidedElsewhere flags).

Ships step 3 of the tracker-schema stretch:

  • Step 1 (#188) — TrackedEntityAttribute + TrackedEntityType.
  • Step 2 (#189) — Program + PTEA + OU scope.
  • Step 3 (this) — ProgramStage + programStageDataElements[].

Surface:

  • create(...) — kwargs over the minimal required subset (name, program_uid) plus common knobs (repeatable, auto_generate_event, min_days_from_start, sort_order, validation_strategy, feature_type).
  • add_element(stage_uid, de_uid, *, compulsory, allow_future_date, display_in_reports, allow_provided_elsewhere, render_options_as_radio) — wire a DE into the stage's data-entry form via the PSDE join table. Round-trips the full stage + PUTs with mergeMode=REPLACE (matches the Program PUT quirk).
  • remove_element / reorder(stage_uid, [de_uids]).
  • rename(uid, ...) / delete(uid) — standard edit pathways.

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

Classes

ProgramStage

Bases: BaseModel

Generated model for DHIS2 ProgramStage.

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

API endpoint: /api/programStages.

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_stage.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
class ProgramStage(BaseModel):
    """Generated model for DHIS2 `ProgramStage`.

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

    API endpoint: /api/programStages.

    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).")
    allowGenerateNextVisit: bool | None = None
    attributeValues: Any | None = Field(default=None, description="Reference to AttributeValues. Length/value max=255.")
    autoGenerateEvent: bool | None = None
    blockEntryForm: bool | None = None
    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.")
    dataEntryForm: Reference | None = Field(default=None, description="Reference to DataEntryForm.")
    description: str | None = Field(default=None, description="Length/value min=2, max=2147483647.")
    displayDescription: str | None = Field(default=None, description="Read-only.")
    displayDueDateLabel: str | None = Field(default=None, description="Read-only.")
    displayEventLabel: str | None = Field(default=None, description="Read-only.")
    displayExecutionDateLabel: str | None = Field(default=None, description="Read-only.")
    displayFormName: str | None = Field(default=None, description="Read-only.")
    displayGenerateEventBox: bool | None = None
    displayName: str | None = Field(default=None, description="Read-only.")
    displayProgramStageLabel: str | None = Field(default=None, description="Read-only.")
    displayShortName: str | None = Field(default=None, description="Read-only.")
    dueDateLabel: str | None = Field(default=None, description="Length/value min=2, max=255.")
    enableUserAssignment: bool | None = None
    eventLabel: str | None = Field(default=None, description="Length/value min=2, max=255.")
    executionDateLabel: str | None = Field(default=None, description="Length/value min=2, max=255.")
    favorite: bool | None = Field(default=None, description="Read-only.")
    favorites: list[Any] | None = Field(default=None, description="Collection of String. Read-only (inverse side).")
    featureType: FeatureType | None = None
    formName: str | None = Field(default=None, description="Length/value max=2147483647.")
    formType: FormType | None = Field(default=None, description="Read-only.")
    generatedByEnrollmentDate: bool | None = None
    hideDueDate: bool | None = None
    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.")
    minDaysFromStart: int | None = Field(default=None, description="Length/value max=2147483647.")
    name: str | None = Field(default=None, description="Length/value min=1, max=230.")
    nextScheduleDate: Reference | None = Field(default=None, description="Reference to DataElement.")
    notificationTemplates: list[Any] | None = Field(
        default=None, description="Collection of ProgramNotificationTemplate."
    )
    openAfterEnrollment: bool | None = None
    periodType: PeriodType | None = Field(default=None, description="Reference to PeriodType. Length/value max=255.")
    preGenerateUID: bool | None = None
    program: Reference | None = Field(default=None, description="Reference to Program.")
    programStageDataElements: list[Any] | None = Field(
        default=None, description="Collection of ProgramStageDataElement."
    )
    programStageLabel: str | None = Field(default=None, description="Length/value min=2, max=255.")
    programStageSections: list[Any] | None = Field(default=None, description="Collection of ProgramStageSection.")
    referral: bool | None = None
    remindCompleted: bool | None = None
    repeatable: bool | None = None
    reportDateToUse: str | None = Field(default=None, description="Length/value max=255.")
    sharing: Any | None = Field(default=None, description="Reference to Sharing. Length/value max=255.")
    shortName: str | None = Field(default=None, description="Length/value min=1, max=50.")
    sortOrder: int | None = Field(default=None, description="Length/value max=2147483647.")
    standardInterval: int | None = Field(default=None, description="Length/value max=2147483647.")
    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).")
    validationStrategy: ValidationStrategy | None = None

ProgramStagesAccessor

Dhis2Client.program_stages — CRUD + PSDE ordering helpers.

Source code in packages/dhis2w-client/src/dhis2w_client/program_stages.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
class ProgramStagesAccessor:
    """`Dhis2Client.program_stages` — CRUD + PSDE ordering helpers."""

    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[ProgramStage]:
        """Page through ProgramStages, optionally filtered to one parent Program."""
        filters: list[str] | None = None
        if program_uid:
            filters = [f"program.id:eq:{program_uid}"]
        return cast(
            list[ProgramStage],
            await self._client.resources.program_stages.list(
                fields=_STAGE_FIELDS,
                filters=filters,
                page=page,
                page_size=page_size,
            ),
        )

    async def list_for(self, program_uid: str) -> list[ProgramStage]:
        """Return ProgramStages belonging to one Program, sorted by `sortOrder`."""
        return cast(
            list[ProgramStage],
            await self._client.resources.program_stages.list(
                fields=_STAGE_FIELDS,
                filters=[f"program.id:eq:{program_uid}"],
                order=["sortOrder:asc"],
                paging=False,
            ),
        )

    async def get(self, uid: str) -> ProgramStage:
        """Fetch one ProgramStage with its PSDE list resolved inline."""
        return await self._client.get(
            f"/api/programStages/{uid}",
            model=ProgramStage,
            params={"fields": _STAGE_FIELDS},
        )

    async def create(
        self,
        *,
        name: str,
        program_uid: str,
        short_name: str | None = None,
        description: str | None = None,
        code: str | None = None,
        form_name: str | None = None,
        sort_order: int | None = None,
        repeatable: bool | None = None,
        auto_generate_event: bool | None = None,
        generated_by_enrollment_date: bool | None = None,
        open_after_enrollment: bool | None = None,
        block_entry_form: bool | None = None,
        feature_type: str | None = None,
        period_type: PeriodType | str | None = None,
        validation_strategy: str | None = None,
        min_days_from_start: int | None = None,
        standard_interval: int | None = None,
        enable_user_assignment: bool | None = None,
        pre_generate_uid: bool | None = None,
        due_date_label: str | None = None,
        execution_date_label: str | None = None,
        event_label: str | None = None,
        uid: str | None = None,
    ) -> ProgramStage:
        """Create a ProgramStage under `program_uid`.

        `repeatable=True` makes the stage reusable within one enrollment
        (follow-up ANC visits, chronic-care check-ins).
        `auto_generate_event=True` + `generated_by_enrollment_date=True`
        tells DHIS2 to schedule an event when the enrollment is
        created. `min_days_from_start` + `standard_interval` tune the
        default due-date math.
        """
        payload: dict[str, Any] = {
            "name": name,
            "program": {"id": program_uid},
        }
        if short_name:
            payload["shortName"] = short_name
        if description:
            payload["description"] = description
        if code:
            payload["code"] = code
        if form_name:
            payload["formName"] = form_name
        if sort_order is not None:
            payload["sortOrder"] = sort_order
        if repeatable is not None:
            payload["repeatable"] = repeatable
        if auto_generate_event is not None:
            payload["autoGenerateEvent"] = auto_generate_event
        if generated_by_enrollment_date is not None:
            payload["generatedByEnrollmentDate"] = generated_by_enrollment_date
        if open_after_enrollment is not None:
            payload["openAfterEnrollment"] = open_after_enrollment
        if block_entry_form is not None:
            payload["blockEntryForm"] = block_entry_form
        if feature_type:
            payload["featureType"] = feature_type
        if period_type is not None:
            payload["periodType"] = period_type.value if isinstance(period_type, PeriodType) else period_type
        if validation_strategy:
            payload["validationStrategy"] = validation_strategy
        if min_days_from_start is not None:
            payload["minDaysFromStart"] = min_days_from_start
        if standard_interval is not None:
            payload["standardInterval"] = standard_interval
        if enable_user_assignment is not None:
            payload["enableUserAssignment"] = enable_user_assignment
        if pre_generate_uid is not None:
            payload["preGenerateUID"] = pre_generate_uid
        if due_date_label:
            payload["dueDateLabel"] = due_date_label
        if execution_date_label:
            payload["executionDateLabel"] = execution_date_label
        if event_label:
            payload["eventLabel"] = event_label
        if uid:
            payload["id"] = uid
        envelope = await self._client.post("/api/programStages", payload, model=WebMessageResponse)
        created_uid = envelope.created_uid or uid
        if not created_uid:
            raise RuntimeError("program-stage create did not return a uid")
        return await self.get(created_uid)

    async def update(self, stage: ProgramStage) -> ProgramStage:
        """PUT an edited ProgramStage back. `stage.id` must be set.

        Matches the Program PUT quirk: `mergeMode=REPLACE` forces
        nested-list replacement so `programStageDataElements[]` items
        omitted from the payload actually get removed instead of
        silently retained.
        """
        if not stage.id:
            raise ValueError("update requires stage.id to be set")
        body = stage.model_dump(by_alias=True, exclude_none=True, mode="json")
        _strip_self_ref_from_psde(body)
        await self._client.put_raw(f"/api/programStages/{stage.id}", body=body, params={"mergeMode": "REPLACE"})
        return await self.get(stage.id)

    async def rename(
        self,
        uid: str,
        *,
        name: str | None = None,
        short_name: str | None = None,
        form_name: str | None = None,
        description: str | None = None,
    ) -> ProgramStage:
        """Partial-update shortcut — read, mutate label fields, PUT."""
        if name is None and short_name is None and form_name is None and description is None:
            raise ValueError("rename requires at least one of name / short_name / form_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 form_name is not None:
            current.formName = form_name
        if description is not None:
            current.description = description
        return await self.update(current)

    async def add_element(
        self,
        stage_uid: str,
        data_element_uid: str,
        *,
        compulsory: bool = False,
        allow_future_date: bool = False,
        display_in_reports: bool = True,
        allow_provided_elsewhere: bool = False,
        render_options_as_radio: bool = False,
        skip_synchronization: bool = False,
        skip_analytics: bool = False,
        sort_order: int | None = None,
    ) -> ProgramStage:
        """Wire a DataElement into the ProgramStage via the PSDE join table.

        Round-trips the full stage, appends a new PSDE entry, PUTs with
        `mergeMode=REPLACE`. Idempotent — returns early when the DE is
        already attached.
        """
        current = await self.get(stage_uid)
        raw = current.model_dump(by_alias=True, exclude_none=True, mode="json")
        _strip_self_ref_from_psde(raw)
        existing = raw.get("programStageDataElements") or []
        if any(
            (entry.get("dataElement") or {}).get("id") == data_element_uid
            for entry in existing
            if isinstance(entry, dict)
        ):
            return current
        new_entry: dict[str, Any] = {
            "dataElement": {"id": data_element_uid},
            "compulsory": compulsory,
            "allowFutureDate": allow_future_date,
            "displayInReports": display_in_reports,
            "allowProvidedElsewhere": allow_provided_elsewhere,
            "renderOptionsAsRadio": render_options_as_radio,
            "skipSynchronization": skip_synchronization,
            "skipAnalytics": skip_analytics,
        }
        if sort_order is not None:
            new_entry["sortOrder"] = sort_order
        existing.append(new_entry)
        raw["programStageDataElements"] = existing
        await self._client.put_raw(f"/api/programStages/{stage_uid}", body=raw, params={"mergeMode": "REPLACE"})
        return await self.get(stage_uid)

    async def remove_element(self, stage_uid: str, data_element_uid: str) -> ProgramStage:
        """Drop a DataElement from the ProgramStage's PSDE list."""
        current = await self.get(stage_uid)
        raw = current.model_dump(by_alias=True, exclude_none=True, mode="json")
        _strip_self_ref_from_psde(raw)
        existing = raw.get("programStageDataElements") or []
        filtered = [
            entry
            for entry in existing
            if isinstance(entry, dict) and (entry.get("dataElement") or {}).get("id") != data_element_uid
        ]
        if len(filtered) == len(existing):
            return current
        raw["programStageDataElements"] = filtered
        await self._client.put_raw(f"/api/programStages/{stage_uid}", body=raw, params={"mergeMode": "REPLACE"})
        return await self.get(stage_uid)

    async def reorder(self, stage_uid: str, data_element_uids: list[str]) -> ProgramStage:
        """Replace the ordered `programStageDataElements` with exactly the given DE UIDs.

        Any PSDE flags (`compulsory`, `display_in_reports`, etc.) on
        dropped entries are lost. Use `add_element` + `remove_element`
        for fine-grained edits; reach for `reorder` when the set of
        attached DEs is known.
        """
        current = await self.get(stage_uid)
        raw = current.model_dump(by_alias=True, exclude_none=True, mode="json")
        _strip_self_ref_from_psde(raw)
        # Preserve PSDE flags for DEs that stay in the list.
        existing = {
            (entry.get("dataElement") or {}).get("id"): entry
            for entry in (raw.get("programStageDataElements") or [])
            if isinstance(entry, dict)
        }
        new_entries: list[dict[str, Any]] = []
        for index, de_uid in enumerate(data_element_uids):
            entry = existing.get(de_uid) or {"dataElement": {"id": de_uid}}
            entry["sortOrder"] = index
            new_entries.append(entry)
        raw["programStageDataElements"] = new_entries
        await self._client.put_raw(f"/api/programStages/{stage_uid}", body=raw, params={"mergeMode": "REPLACE"})
        return await self.get(stage_uid)

    async def delete(self, uid: str) -> None:
        """Delete a ProgramStage — DHIS2 rejects deletes on stages with recorded events."""
        if not uid:
            raise ValueError("delete requires a non-empty uid")
        await self._client.resources.program_stages.delete(uid)
Functions
__init__(client)

Bind to the sharing client.

Source code in packages/dhis2w-client/src/dhis2w_client/program_stages.py 
65
66
67
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 ProgramStages, optionally filtered to one parent Program.

Source code in packages/dhis2w-client/src/dhis2w_client/program_stages.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
async def list_all(
    self,
    *,
    program_uid: str | None = None,
    page: int = 1,
    page_size: int = 50,
) -> list[ProgramStage]:
    """Page through ProgramStages, optionally filtered to one parent Program."""
    filters: list[str] | None = None
    if program_uid:
        filters = [f"program.id:eq:{program_uid}"]
    return cast(
        list[ProgramStage],
        await self._client.resources.program_stages.list(
            fields=_STAGE_FIELDS,
            filters=filters,
            page=page,
            page_size=page_size,
        ),
    )
list_for(program_uid) async

Return ProgramStages belonging to one Program, sorted by sortOrder.

Source code in packages/dhis2w-client/src/dhis2w_client/program_stages.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
async def list_for(self, program_uid: str) -> list[ProgramStage]:
    """Return ProgramStages belonging to one Program, sorted by `sortOrder`."""
    return cast(
        list[ProgramStage],
        await self._client.resources.program_stages.list(
            fields=_STAGE_FIELDS,
            filters=[f"program.id:eq:{program_uid}"],
            order=["sortOrder:asc"],
            paging=False,
        ),
    )
get(uid) async

Fetch one ProgramStage with its PSDE list resolved inline.

Source code in packages/dhis2w-client/src/dhis2w_client/program_stages.py 
102
103
104
105
106
107
108
async def get(self, uid: str) -> ProgramStage:
    """Fetch one ProgramStage with its PSDE list resolved inline."""
    return await self._client.get(
        f"/api/programStages/{uid}",
        model=ProgramStage,
        params={"fields": _STAGE_FIELDS},
    )
create(*, name, program_uid, short_name=None, description=None, code=None, form_name=None, sort_order=None, repeatable=None, auto_generate_event=None, generated_by_enrollment_date=None, open_after_enrollment=None, block_entry_form=None, feature_type=None, period_type=None, validation_strategy=None, min_days_from_start=None, standard_interval=None, enable_user_assignment=None, pre_generate_uid=None, due_date_label=None, execution_date_label=None, event_label=None, uid=None) async

Create a ProgramStage under program_uid.

repeatable=True makes the stage reusable within one enrollment (follow-up ANC visits, chronic-care check-ins). auto_generate_event=True + generated_by_enrollment_date=True tells DHIS2 to schedule an event when the enrollment is created. min_days_from_start + standard_interval tune the default due-date math.

Source code in packages/dhis2w-client/src/dhis2w_client/program_stages.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
async def create(
    self,
    *,
    name: str,
    program_uid: str,
    short_name: str | None = None,
    description: str | None = None,
    code: str | None = None,
    form_name: str | None = None,
    sort_order: int | None = None,
    repeatable: bool | None = None,
    auto_generate_event: bool | None = None,
    generated_by_enrollment_date: bool | None = None,
    open_after_enrollment: bool | None = None,
    block_entry_form: bool | None = None,
    feature_type: str | None = None,
    period_type: PeriodType | str | None = None,
    validation_strategy: str | None = None,
    min_days_from_start: int | None = None,
    standard_interval: int | None = None,
    enable_user_assignment: bool | None = None,
    pre_generate_uid: bool | None = None,
    due_date_label: str | None = None,
    execution_date_label: str | None = None,
    event_label: str | None = None,
    uid: str | None = None,
) -> ProgramStage:
    """Create a ProgramStage under `program_uid`.

    `repeatable=True` makes the stage reusable within one enrollment
    (follow-up ANC visits, chronic-care check-ins).
    `auto_generate_event=True` + `generated_by_enrollment_date=True`
    tells DHIS2 to schedule an event when the enrollment is
    created. `min_days_from_start` + `standard_interval` tune the
    default due-date math.
    """
    payload: dict[str, Any] = {
        "name": name,
        "program": {"id": program_uid},
    }
    if short_name:
        payload["shortName"] = short_name
    if description:
        payload["description"] = description
    if code:
        payload["code"] = code
    if form_name:
        payload["formName"] = form_name
    if sort_order is not None:
        payload["sortOrder"] = sort_order
    if repeatable is not None:
        payload["repeatable"] = repeatable
    if auto_generate_event is not None:
        payload["autoGenerateEvent"] = auto_generate_event
    if generated_by_enrollment_date is not None:
        payload["generatedByEnrollmentDate"] = generated_by_enrollment_date
    if open_after_enrollment is not None:
        payload["openAfterEnrollment"] = open_after_enrollment
    if block_entry_form is not None:
        payload["blockEntryForm"] = block_entry_form
    if feature_type:
        payload["featureType"] = feature_type
    if period_type is not None:
        payload["periodType"] = period_type.value if isinstance(period_type, PeriodType) else period_type
    if validation_strategy:
        payload["validationStrategy"] = validation_strategy
    if min_days_from_start is not None:
        payload["minDaysFromStart"] = min_days_from_start
    if standard_interval is not None:
        payload["standardInterval"] = standard_interval
    if enable_user_assignment is not None:
        payload["enableUserAssignment"] = enable_user_assignment
    if pre_generate_uid is not None:
        payload["preGenerateUID"] = pre_generate_uid
    if due_date_label:
        payload["dueDateLabel"] = due_date_label
    if execution_date_label:
        payload["executionDateLabel"] = execution_date_label
    if event_label:
        payload["eventLabel"] = event_label
    if uid:
        payload["id"] = uid
    envelope = await self._client.post("/api/programStages", payload, model=WebMessageResponse)
    created_uid = envelope.created_uid or uid
    if not created_uid:
        raise RuntimeError("program-stage create did not return a uid")
    return await self.get(created_uid)
update(stage) async

PUT an edited ProgramStage back. stage.id must be set.

Matches the Program PUT quirk: mergeMode=REPLACE forces nested-list replacement so programStageDataElements[] items omitted from the payload actually get removed instead of silently retained.

Source code in packages/dhis2w-client/src/dhis2w_client/program_stages.py 
198
199
200
201
202
203
204
205
206
207
208
209
210
211
async def update(self, stage: ProgramStage) -> ProgramStage:
    """PUT an edited ProgramStage back. `stage.id` must be set.

    Matches the Program PUT quirk: `mergeMode=REPLACE` forces
    nested-list replacement so `programStageDataElements[]` items
    omitted from the payload actually get removed instead of
    silently retained.
    """
    if not stage.id:
        raise ValueError("update requires stage.id to be set")
    body = stage.model_dump(by_alias=True, exclude_none=True, mode="json")
    _strip_self_ref_from_psde(body)
    await self._client.put_raw(f"/api/programStages/{stage.id}", body=body, params={"mergeMode": "REPLACE"})
    return await self.get(stage.id)
rename(uid, *, name=None, short_name=None, form_name=None, description=None) async

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

Source code in packages/dhis2w-client/src/dhis2w_client/program_stages.py 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
async def rename(
    self,
    uid: str,
    *,
    name: str | None = None,
    short_name: str | None = None,
    form_name: str | None = None,
    description: str | None = None,
) -> ProgramStage:
    """Partial-update shortcut — read, mutate label fields, PUT."""
    if name is None and short_name is None and form_name is None and description is None:
        raise ValueError("rename requires at least one of name / short_name / form_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 form_name is not None:
        current.formName = form_name
    if description is not None:
        current.description = description
    return await self.update(current)
add_element(stage_uid, data_element_uid, *, compulsory=False, allow_future_date=False, display_in_reports=True, allow_provided_elsewhere=False, render_options_as_radio=False, skip_synchronization=False, skip_analytics=False, sort_order=None) async

Wire a DataElement into the ProgramStage via the PSDE join table.

Round-trips the full stage, appends a new PSDE entry, PUTs with mergeMode=REPLACE. Idempotent — returns early when the DE is already attached.

Source code in packages/dhis2w-client/src/dhis2w_client/program_stages.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
async def add_element(
    self,
    stage_uid: str,
    data_element_uid: str,
    *,
    compulsory: bool = False,
    allow_future_date: bool = False,
    display_in_reports: bool = True,
    allow_provided_elsewhere: bool = False,
    render_options_as_radio: bool = False,
    skip_synchronization: bool = False,
    skip_analytics: bool = False,
    sort_order: int | None = None,
) -> ProgramStage:
    """Wire a DataElement into the ProgramStage via the PSDE join table.

    Round-trips the full stage, appends a new PSDE entry, PUTs with
    `mergeMode=REPLACE`. Idempotent — returns early when the DE is
    already attached.
    """
    current = await self.get(stage_uid)
    raw = current.model_dump(by_alias=True, exclude_none=True, mode="json")
    _strip_self_ref_from_psde(raw)
    existing = raw.get("programStageDataElements") or []
    if any(
        (entry.get("dataElement") or {}).get("id") == data_element_uid
        for entry in existing
        if isinstance(entry, dict)
    ):
        return current
    new_entry: dict[str, Any] = {
        "dataElement": {"id": data_element_uid},
        "compulsory": compulsory,
        "allowFutureDate": allow_future_date,
        "displayInReports": display_in_reports,
        "allowProvidedElsewhere": allow_provided_elsewhere,
        "renderOptionsAsRadio": render_options_as_radio,
        "skipSynchronization": skip_synchronization,
        "skipAnalytics": skip_analytics,
    }
    if sort_order is not None:
        new_entry["sortOrder"] = sort_order
    existing.append(new_entry)
    raw["programStageDataElements"] = existing
    await self._client.put_raw(f"/api/programStages/{stage_uid}", body=raw, params={"mergeMode": "REPLACE"})
    return await self.get(stage_uid)
remove_element(stage_uid, data_element_uid) async

Drop a DataElement from the ProgramStage's PSDE list.

Source code in packages/dhis2w-client/src/dhis2w_client/program_stages.py 
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
async def remove_element(self, stage_uid: str, data_element_uid: str) -> ProgramStage:
    """Drop a DataElement from the ProgramStage's PSDE list."""
    current = await self.get(stage_uid)
    raw = current.model_dump(by_alias=True, exclude_none=True, mode="json")
    _strip_self_ref_from_psde(raw)
    existing = raw.get("programStageDataElements") or []
    filtered = [
        entry
        for entry in existing
        if isinstance(entry, dict) and (entry.get("dataElement") or {}).get("id") != data_element_uid
    ]
    if len(filtered) == len(existing):
        return current
    raw["programStageDataElements"] = filtered
    await self._client.put_raw(f"/api/programStages/{stage_uid}", body=raw, params={"mergeMode": "REPLACE"})
    return await self.get(stage_uid)
reorder(stage_uid, data_element_uids) async

Replace the ordered programStageDataElements with exactly the given DE UIDs.

Any PSDE flags (compulsory, display_in_reports, etc.) on dropped entries are lost. Use add_element + remove_element for fine-grained edits; reach for reorder when the set of attached DEs is known.

Source code in packages/dhis2w-client/src/dhis2w_client/program_stages.py 
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
async def reorder(self, stage_uid: str, data_element_uids: list[str]) -> ProgramStage:
    """Replace the ordered `programStageDataElements` with exactly the given DE UIDs.

    Any PSDE flags (`compulsory`, `display_in_reports`, etc.) on
    dropped entries are lost. Use `add_element` + `remove_element`
    for fine-grained edits; reach for `reorder` when the set of
    attached DEs is known.
    """
    current = await self.get(stage_uid)
    raw = current.model_dump(by_alias=True, exclude_none=True, mode="json")
    _strip_self_ref_from_psde(raw)
    # Preserve PSDE flags for DEs that stay in the list.
    existing = {
        (entry.get("dataElement") or {}).get("id"): entry
        for entry in (raw.get("programStageDataElements") or [])
        if isinstance(entry, dict)
    }
    new_entries: list[dict[str, Any]] = []
    for index, de_uid in enumerate(data_element_uids):
        entry = existing.get(de_uid) or {"dataElement": {"id": de_uid}}
        entry["sortOrder"] = index
        new_entries.append(entry)
    raw["programStageDataElements"] = new_entries
    await self._client.put_raw(f"/api/programStages/{stage_uid}", body=raw, params={"mergeMode": "REPLACE"})
    return await self.get(stage_uid)
delete(uid) async

Delete a ProgramStage — DHIS2 rejects deletes on stages with recorded events.

Source code in packages/dhis2w-client/src/dhis2w_client/program_stages.py 
326
327
328
329
330
async def delete(self, uid: str) -> None:
    """Delete a ProgramStage — DHIS2 rejects deletes on stages with recorded events."""
    if not uid:
        raise ValueError("delete requires a non-empty uid")
    await self._client.resources.program_stages.delete(uid)