Skip to content

Organisation units

Four accessors on Dhis2Client cover the DHIS2 X / XGroup / XGroupSet triple for the organisation-unit hierarchy, plus per-level naming:

Accessor API path Purpose
client.organisation_units /api/organisationUnits Hierarchy nodes — parent / children / descendants / move / create / delete
client.organisation_unit_groups /api/organisationUnitGroups Orthogonal groupings (public/private, urban/rural, …) + member add/remove
client.organisation_unit_group_sets /api/organisationUnitGroupSets Analytics dimensions that collect groups (e.g. "Facility Ownership")
client.organisation_unit_levels /api/organisationUnitLevels Per-depth naming — give level 2 a human label like "Province"

Generic CRUD is still available on the generated accessors (client.resources.organisation_units and friends). The hand-written accessors layer workflow primitives — bounded subtree walks, per-item membership POST/DELETE shortcuts, rename_by_level — that the generated CRUD can't express in one call.

Naming convention

The CLI, MCP tools, and Python attribute names mirror the DHIS2 API paths 1:1 (camelCase → kebab-case / snake_case):

DHIS2 CLI MCP tool prefix Client attribute
/api/organisationUnits dhis2 metadata organisation-units … metadata_organisation_unit_* client.organisation_units
/api/organisationUnitGroups dhis2 metadata organisation-unit-groups … metadata_organisation_unit_group_* client.organisation_unit_groups
/api/organisationUnitGroupSets dhis2 metadata organisation-unit-group-sets … metadata_organisation_unit_group_set_* client.organisation_unit_group_sets
/api/organisationUnitLevels dhis2 metadata organisation-unit-levels … metadata_organisation_unit_level_* client.organisation_unit_levels

One rule — "lowercase the DHIS2 resource name, hyphenate or underscore the camelCase boundary" — so anyone who knows the API URL can guess every other surface. The same rule will apply to every future X / XGroup / XGroupSet triple (data elements, indicators, category options, program indicators).

No *Spec builders

Unlike the legend-set / map / visualisation accessors, the organisation-unit accessors don't ship dedicated *Spec classes. The argument is that OU + group + level creates are narrow enough (5–10 meaningful fields) that keyword args on the accessor are cleaner than a spec-over-model hop. Callers that need fine-grained control over the full generated OrganisationUnit model still have update(unit) — mutate the returned model, pass it back.

This sits on the kwargs side of the resolved spec-vs-kwargs rule documented on the Legend sets doc: a *Spec is justified only where the wire shape needs transformation the generated model can't carry (chart-type-aware placement, enum fan-out, inline children with synthesised UIDs). OUs need none of that, so kwargs win.

Tree navigation

async with Dhis2Client(...) as client:
    roots = await client.organisation_units.list_by_level(1)
    subtree = await client.organisation_units.list_descendants(
        roots[0].id,
        max_depth=2,
    )
    for unit in subtree:
        print(f"L{unit.hierarchyLevel}  {unit.name}  ({unit.id})")

list_descendants is breadth-first — the returned list opens with the root, then every direct child, then grandchildren. max_depth=0 returns just the root.

Group + group-set membership

group = await client.organisation_unit_groups.get("CXw2yu5fodb")
members = await client.organisation_unit_groups.list_members(group.id, page_size=200)

# Move five facilities into a different grouping without a full PUT
await client.organisation_unit_groups.add_members(
    "CXw2yu5fodb",
    ou_uids=[f.id for f in members[:5]],
)

# Wire the group into a new analytics dimension
group_set = await client.organisation_unit_group_sets.create(
    name="Provincial oversight",
    short_name="ProvOversight",
)
await client.organisation_unit_group_sets.add_groups(
    group_set.id,
    group_uids=[group.id],
)

Both add_members / add_groups use the per-item POST /<resource>/{uid}/<inverse>/{target_uid} DHIS2 shortcut so the server never sees a full-collection payload. That keeps edits atomic and avoids the "losing members on round-trip" class of bug where a naive PUT drops rows that changed under you.

Naming levels

DHIS2 auto-creates one OrganisationUnitLevel row per depth the first time an OU is stored at that level. They arrive without human labels — renaming them to Country, Province, District, Facility is the main write operation:

await client.organisation_unit_levels.rename_by_level(2, name="Province")
await client.organisation_unit_levels.rename_by_level(3, name="District")
await client.organisation_unit_levels.rename_by_level(4, name="Facility")

The CLI mirrors this:

dhis2 metadata organisation-unit-levels list
dhis2 metadata organisation-unit-levels rename 2 --by-level --name Province

organisation_units

OrganisationUnit authoring + tree navigation — Dhis2Client.organisation_units.

DHIS2 models its geographic hierarchy as a single tree of OrganisationUnit rows: every non-root unit carries a parent reference + a materialised path string (/<root>/<child>/...). Generic CRUD lives on the generated accessor (client.resources.organisation_units); this module adds the workflow primitives the tree shape demands but the bare CRUD can't express in one call:

  • Walk a subtree at a bounded depth (list_descendants) without fetching every OU on the instance.
  • Enumerate OUs at a given level across the whole tree.
  • Reparent a unit via move(uid, new_parent_uid) — turns into a full PUT of the OU with the parent ref swapped; DHIS2 rebuilds path server-side on save.
  • create_under(parent_uid, ...) — fills in the parent + opening-date defaults so callers don't hand-roll the reference payload.

All reads return typed OrganisationUnit instances; writes go through the raw client helpers so the caller sees DHIS2's computed fields (path, hierarchyLevel, leaf, …) on the returned model.

Classes

OrganisationUnit

Bases: BaseModel

Generated model for DHIS2 OrganisationUnit.

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

API endpoint: /api/organisationUnits.

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

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

    API endpoint: /api/organisationUnits.

    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).")
    address: str | None = Field(default=None, description="Length/value max=255.")
    aggregationType: AggregationType | None = None
    attributeValues: Any | None = Field(default=None, description="Reference to AttributeValues. Length/value max=255.")
    childs: list[Any] | None = Field(
        default=None, description="Collection of OrganisationUnit. Read-only (inverse side)."
    )
    closedDate: datetime | None = None
    code: str | None = Field(default=None, description="Unique. Length/value max=50.")
    comment: str | None = Field(default=None, description="Length/value max=2147483647.")
    contactPerson: str | None = Field(default=None, description="Length/value max=255.")
    created: datetime | None = None
    createdBy: Reference | None = Field(default=None, description="Reference to User.")
    dataSets: list[Any] | None = Field(default=None, description="Collection of DataSet. Read-only (inverse side).")
    description: str | None = Field(default=None, description="Length/value min=1, max=2147483647.")
    dimensionItem: str | None = Field(default=None, description="Read-only.")
    dimensionItemType: DimensionItemType | None = None
    displayDescription: str | None = Field(default=None, description="Read-only.")
    displayFormName: str | None = Field(default=None, description="Read-only.")
    displayName: str | None = Field(default=None, description="Read-only.")
    displayShortName: str | None = Field(default=None, description="Read-only.")
    email: str | None = Field(default=None, description="Length/value max=150.")
    favorite: bool | None = Field(default=None, description="Read-only.")
    favorites: list[Any] | None = Field(default=None, description="Collection of String. Read-only (inverse side).")
    formName: str | None = Field(default=None, description="Length/value max=2147483647.")
    geometry: Any | None = Field(default=None, description="Reference to Geometry. Length/value max=131072.")
    href: str | None = None
    id: str | None = Field(default=None, description="Unique. Length/value min=11, max=11.")
    image: Reference | None = Field(default=None, description="Reference to FileResource.")
    lastUpdated: datetime | None = None
    lastUpdatedBy: Reference | None = Field(default=None, description="Reference to User.")
    leaf: bool | None = Field(default=None, description="Read-only.")
    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. Read-only (inverse side).")
    level: int | None = Field(default=None, description="Length/value max=2147483647.")
    memberCount: int | None = Field(default=None, description="Length/value max=2147483647.")
    name: str | None = Field(default=None, description="Length/value min=1, max=230.")
    openingDate: datetime | None = None
    organisationUnitGroups: list[Any] | None = Field(
        default=None, description="Collection of OrganisationUnitGroup. Read-only (inverse side)."
    )
    organisationUnits: list[Any] | None = Field(
        default=None, description="Collection of OrganisationUnit. Read-only (inverse side)."
    )
    parent: Reference | None = Field(default=None, description="Reference to OrganisationUnit.")
    path: str | None = Field(default=None, description="Unique. Length/value max=255.")
    phoneNumber: str | None = Field(default=None, description="Length/value max=150.")
    programs: list[Any] | None = Field(default=None, description="Collection of Program. Read-only (inverse side).")
    queryMods: Any | None = Field(default=None, description="Reference to QueryModifiers. Read-only (inverse side).")
    sharing: Any | None = Field(default=None, description="Reference to Sharing. Read-only (inverse side).")
    shortName: str | None = Field(default=None, description="Length/value min=1, max=50.")
    translations: list[Any] | None = Field(default=None, description="Collection of Translation. Length/value max=255.")
    type: str | None = Field(default=None, description="Length/value max=2147483647.")
    url: str | None = Field(default=None, description="Length/value max=255.")
    user: Reference | None = Field(default=None, description="Reference to User. Read-only (inverse side).")
    userItems: list[Any] | None = Field(default=None, description="Collection of User. Read-only (inverse side).")

OrganisationUnitsAccessor

Dhis2Client.organisation_units — tree-aware reads + authoring helpers.

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_units.py
class OrganisationUnitsAccessor:
    """`Dhis2Client.organisation_units` — tree-aware reads + authoring helpers."""

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

    async def list_all(
        self,
        *,
        level: int | None = None,
        page: int = 1,
        page_size: int = 50,
    ) -> list[OrganisationUnit]:
        """List OUs page-by-page, optionally filtered to one `level`.

        Use `level=1` to enumerate top-level roots; higher levels for
        countries / regions / facilities depending on the instance's
        hierarchy depth. Paging is server-side — callers that need every
        row should loop over `page` until the returned list is shorter
        than `page_size`.
        """
        filters: list[str] | None = None
        if level is not None:
            filters = [f"level:eq:{level}"]
        return cast(
            list[OrganisationUnit],
            await self._client.resources.organisation_units.list(
                fields=_OU_FIELDS,
                filters=filters,
                page=page,
                page_size=page_size,
            ),
        )

    async def get(self, uid: str) -> OrganisationUnit:
        """Fetch one OU by UID with parent + hierarchy fields populated."""
        return await self._client.get(
            f"/api/organisationUnits/{uid}", model=OrganisationUnit, params={"fields": _OU_FIELDS}
        )

    async def list_children(self, parent_uid: str) -> list[OrganisationUnit]:
        """Direct children of one OU, sorted by name.

        One-level walk — used by CLI tree renderers and by analytics
        pickers that show a lazy drill-down. For the whole subtree use
        `list_descendants` with a bounded depth.
        """
        return cast(
            list[OrganisationUnit],
            await self._client.resources.organisation_units.list(
                fields=_OU_FIELDS,
                filters=[f"parent.id:eq:{parent_uid}"],
                order=["name:asc"],
                paging=False,
            ),
        )

    async def list_descendants(
        self,
        root_uid: str,
        *,
        max_depth: int = 3,
    ) -> list[OrganisationUnit]:
        """Walk a subtree at a bounded depth; include the root itself.

        Breadth-first: emits the root, then every direct child, then
        grandchildren, up to `max_depth` extra levels. Callers rendering
        a tree can group by `hierarchyLevel` to get canonical
        top-to-bottom ordering.
        """
        if max_depth < 0:
            raise ValueError("max_depth must be >= 0")
        root = await self.get(root_uid)
        collected: list[OrganisationUnit] = [root]
        frontier = [root_uid]
        for _depth in range(max_depth):
            if not frontier:
                break
            next_frontier: list[str] = []
            for parent_uid in frontier:
                children = await self.list_children(parent_uid)
                for child in children:
                    collected.append(child)
                    if child.id:
                        next_frontier.append(child.id)
            frontier = next_frontier
        return collected

    async def list_by_level(self, level: int) -> list[OrganisationUnit]:
        """Every OU at a given level — all provinces, all districts, etc.

        Convenience over `list_all(level=..., paging=false)`; sorts by
        `path` so parent/child ordering is stable for reports.
        """
        return cast(
            list[OrganisationUnit],
            await self._client.resources.organisation_units.list(
                fields=_OU_FIELDS,
                filters=[f"level:eq:{level}"],
                order=["path:asc"],
                paging=False,
            ),
        )

    async def create_under(
        self,
        parent_uid: str,
        *,
        name: str,
        short_name: str,
        opening_date: datetime | str,
        uid: str | None = None,
        code: str | None = None,
        description: str | None = None,
    ) -> OrganisationUnit:
        """Create an OU as a child of `parent_uid`.

        DHIS2 requires `openingDate` on every OU (even an ISO-8601 date
        string works). `short_name` must be <=50 chars; DHIS2 rejects
        longer values. Returns the freshly-fetched OU so the caller sees
        `path` + `hierarchyLevel` populated by the server.
        """
        payload: dict[str, Any] = {
            "name": name,
            "shortName": short_name,
            "openingDate": _serialise_date(opening_date),
            "parent": {"id": parent_uid},
        }
        if uid:
            payload["id"] = uid
        if code:
            payload["code"] = code
        if description:
            payload["description"] = description
        created = await self._client.post("/api/organisationUnits", payload, model=WebMessageResponse)
        created_uid = created.created_uid or uid
        if not created_uid:
            raise RuntimeError("organisation-unit create did not return a uid")
        return await self.get(created_uid)

    async def update(self, unit: OrganisationUnit) -> OrganisationUnit:
        """PUT an edited OU back — use after mutating `name`, `description`, etc.

        `unit.id` must be set. DHIS2 recomputes `path` + `hierarchyLevel`
        on the server so the returned model is the authoritative view.
        """
        if not unit.id:
            raise ValueError("update requires unit.id to be set")
        body = unit.model_dump(by_alias=True, exclude_none=True, mode="json")
        await self._client.put_raw(f"/api/organisationUnits/{unit.id}", body=body)
        return await self.get(unit.id)

    async def move(self, uid: str, new_parent_uid: str) -> OrganisationUnit:
        """Reparent one OU. DHIS2 rebuilds `path` + `hierarchyLevel` server-side.

        Implemented as a full PUT of the OU with `parent` swapped — the
        JSON-Patch path on DHIS2 v42 doesn't update the cached
        `path`/`hierarchyLevel` derived columns, so a full PUT is the
        safer shape.
        """
        unit = await self.get(uid)
        unit.parent = Reference(id=new_parent_uid)
        return await self.update(unit)

    async def delete(self, uid: str) -> None:
        """Delete an OU. DHIS2 rejects deletes on units with children or data."""
        if not uid:
            raise ValueError("delete requires a non-empty uid")
        await self._client.resources.organisation_units.delete(uid)
Functions
__init__(client)

Bind to the sharing client.

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

List OUs page-by-page, optionally filtered to one level.

Use level=1 to enumerate top-level roots; higher levels for countries / regions / facilities depending on the instance's hierarchy depth. Paging is server-side — callers that need every row should loop over page until the returned list is shorter than page_size.

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_units.py
async def list_all(
    self,
    *,
    level: int | None = None,
    page: int = 1,
    page_size: int = 50,
) -> list[OrganisationUnit]:
    """List OUs page-by-page, optionally filtered to one `level`.

    Use `level=1` to enumerate top-level roots; higher levels for
    countries / regions / facilities depending on the instance's
    hierarchy depth. Paging is server-side — callers that need every
    row should loop over `page` until the returned list is shorter
    than `page_size`.
    """
    filters: list[str] | None = None
    if level is not None:
        filters = [f"level:eq:{level}"]
    return cast(
        list[OrganisationUnit],
        await self._client.resources.organisation_units.list(
            fields=_OU_FIELDS,
            filters=filters,
            page=page,
            page_size=page_size,
        ),
    )
get(uid) async

Fetch one OU by UID with parent + hierarchy fields populated.

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_units.py
async def get(self, uid: str) -> OrganisationUnit:
    """Fetch one OU by UID with parent + hierarchy fields populated."""
    return await self._client.get(
        f"/api/organisationUnits/{uid}", model=OrganisationUnit, params={"fields": _OU_FIELDS}
    )
list_children(parent_uid) async

Direct children of one OU, sorted by name.

One-level walk — used by CLI tree renderers and by analytics pickers that show a lazy drill-down. For the whole subtree use list_descendants with a bounded depth.

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_units.py
async def list_children(self, parent_uid: str) -> list[OrganisationUnit]:
    """Direct children of one OU, sorted by name.

    One-level walk — used by CLI tree renderers and by analytics
    pickers that show a lazy drill-down. For the whole subtree use
    `list_descendants` with a bounded depth.
    """
    return cast(
        list[OrganisationUnit],
        await self._client.resources.organisation_units.list(
            fields=_OU_FIELDS,
            filters=[f"parent.id:eq:{parent_uid}"],
            order=["name:asc"],
            paging=False,
        ),
    )
list_descendants(root_uid, *, max_depth=3) async

Walk a subtree at a bounded depth; include the root itself.

Breadth-first: emits the root, then every direct child, then grandchildren, up to max_depth extra levels. Callers rendering a tree can group by hierarchyLevel to get canonical top-to-bottom ordering.

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_units.py
async def list_descendants(
    self,
    root_uid: str,
    *,
    max_depth: int = 3,
) -> list[OrganisationUnit]:
    """Walk a subtree at a bounded depth; include the root itself.

    Breadth-first: emits the root, then every direct child, then
    grandchildren, up to `max_depth` extra levels. Callers rendering
    a tree can group by `hierarchyLevel` to get canonical
    top-to-bottom ordering.
    """
    if max_depth < 0:
        raise ValueError("max_depth must be >= 0")
    root = await self.get(root_uid)
    collected: list[OrganisationUnit] = [root]
    frontier = [root_uid]
    for _depth in range(max_depth):
        if not frontier:
            break
        next_frontier: list[str] = []
        for parent_uid in frontier:
            children = await self.list_children(parent_uid)
            for child in children:
                collected.append(child)
                if child.id:
                    next_frontier.append(child.id)
        frontier = next_frontier
    return collected
list_by_level(level) async

Every OU at a given level — all provinces, all districts, etc.

Convenience over list_all(level=..., paging=false); sorts by path so parent/child ordering is stable for reports.

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_units.py
async def list_by_level(self, level: int) -> list[OrganisationUnit]:
    """Every OU at a given level — all provinces, all districts, etc.

    Convenience over `list_all(level=..., paging=false)`; sorts by
    `path` so parent/child ordering is stable for reports.
    """
    return cast(
        list[OrganisationUnit],
        await self._client.resources.organisation_units.list(
            fields=_OU_FIELDS,
            filters=[f"level:eq:{level}"],
            order=["path:asc"],
            paging=False,
        ),
    )
create_under(parent_uid, *, name, short_name, opening_date, uid=None, code=None, description=None) async

Create an OU as a child of parent_uid.

DHIS2 requires openingDate on every OU (even an ISO-8601 date string works). short_name must be <=50 chars; DHIS2 rejects longer values. Returns the freshly-fetched OU so the caller sees path + hierarchyLevel populated by the server.

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_units.py
async def create_under(
    self,
    parent_uid: str,
    *,
    name: str,
    short_name: str,
    opening_date: datetime | str,
    uid: str | None = None,
    code: str | None = None,
    description: str | None = None,
) -> OrganisationUnit:
    """Create an OU as a child of `parent_uid`.

    DHIS2 requires `openingDate` on every OU (even an ISO-8601 date
    string works). `short_name` must be <=50 chars; DHIS2 rejects
    longer values. Returns the freshly-fetched OU so the caller sees
    `path` + `hierarchyLevel` populated by the server.
    """
    payload: dict[str, Any] = {
        "name": name,
        "shortName": short_name,
        "openingDate": _serialise_date(opening_date),
        "parent": {"id": parent_uid},
    }
    if uid:
        payload["id"] = uid
    if code:
        payload["code"] = code
    if description:
        payload["description"] = description
    created = await self._client.post("/api/organisationUnits", payload, model=WebMessageResponse)
    created_uid = created.created_uid or uid
    if not created_uid:
        raise RuntimeError("organisation-unit create did not return a uid")
    return await self.get(created_uid)
update(unit) async

PUT an edited OU back — use after mutating name, description, etc.

unit.id must be set. DHIS2 recomputes path + hierarchyLevel on the server so the returned model is the authoritative view.

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_units.py
async def update(self, unit: OrganisationUnit) -> OrganisationUnit:
    """PUT an edited OU back — use after mutating `name`, `description`, etc.

    `unit.id` must be set. DHIS2 recomputes `path` + `hierarchyLevel`
    on the server so the returned model is the authoritative view.
    """
    if not unit.id:
        raise ValueError("update requires unit.id to be set")
    body = unit.model_dump(by_alias=True, exclude_none=True, mode="json")
    await self._client.put_raw(f"/api/organisationUnits/{unit.id}", body=body)
    return await self.get(unit.id)
move(uid, new_parent_uid) async

Reparent one OU. DHIS2 rebuilds path + hierarchyLevel server-side.

Implemented as a full PUT of the OU with parent swapped — the JSON-Patch path on DHIS2 v42 doesn't update the cached path/hierarchyLevel derived columns, so a full PUT is the safer shape.

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_units.py
async def move(self, uid: str, new_parent_uid: str) -> OrganisationUnit:
    """Reparent one OU. DHIS2 rebuilds `path` + `hierarchyLevel` server-side.

    Implemented as a full PUT of the OU with `parent` swapped — the
    JSON-Patch path on DHIS2 v42 doesn't update the cached
    `path`/`hierarchyLevel` derived columns, so a full PUT is the
    safer shape.
    """
    unit = await self.get(uid)
    unit.parent = Reference(id=new_parent_uid)
    return await self.update(unit)
delete(uid) async

Delete an OU. DHIS2 rejects deletes on units with children or data.

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_units.py
async def delete(self, uid: str) -> None:
    """Delete an OU. DHIS2 rejects deletes on units with children or data."""
    if not uid:
        raise ValueError("delete requires a non-empty uid")
    await self._client.resources.organisation_units.delete(uid)

organisation_unit_groups

OrganisationUnitGroup authoring — Dhis2Client.organisation_unit_groups.

DHIS2 groups organisation units by any orthogonal axis — ownership (public/private), type (urban/rural), program participation — and those groups are what analytics queries pivot on when you ask for a breakdown by "facility type". This accessor wraps the generated CRUD with the membership primitives integration code typically reaches for:

  • list_members — page through the OUs in one group (the group-side inverse of OrganisationUnit.organisationUnitGroups).
  • add_members / remove_members — set-diff style membership edits without hand-rolling the full group payload.

Classes

OrganisationUnitGroup

Bases: BaseModel

Generated model for DHIS2 OrganisationUnitGroup.

DHIS2 Organisation Unit Group - persisted metadata (generated from /api/schemas at DHIS2 v42).

API endpoint: /api/organisationUnitGroups.

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

    DHIS2 Organisation Unit Group - persisted metadata (generated from /api/schemas at DHIS2 v42).

    API endpoint: /api/organisationUnitGroups.

    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.")
    color: str | None = Field(default=None, description="Length/value max=255.")
    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.")
    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.")
    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 = Field(default=None, description="Read-only.")
    formName: str | None = Field(default=None, description="Length/value max=2147483647.")
    geometry: Any | None = Field(default=None, description="Reference to Geometry. Length/value max=255.")
    groupSets: list[Any] | None = Field(
        default=None, description="Collection of OrganisationUnitGroupSet. Read-only (inverse side)."
    )
    href: str | None = None
    id: str | None = Field(default=None, description="Unique. Length/value min=11, max=11.")
    lastUpdated: datetime | None = None
    lastUpdatedBy: Reference | None = Field(default=None, description="Reference to User.")
    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. Read-only (inverse side).")
    name: str | None = Field(default=None, description="Unique. Length/value min=1, max=230.")
    organisationUnits: list[Any] | None = Field(default=None, description="Collection of OrganisationUnit.")
    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.")
    symbol: str | None = Field(default=None, description="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).")

OrganisationUnitGroupsAccessor

Dhis2Client.organisation_unit_groups — CRUD + membership helpers.

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

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

    async def list_all(self) -> list[OrganisationUnitGroup]:
        """Return every OrganisationUnitGroup with its member refs inline."""
        return cast(
            list[OrganisationUnitGroup],
            await self._client.resources.organisation_unit_groups.list(
                fields=_OU_GROUP_FIELDS,
                paging=False,
            ),
        )

    async def get(self, uid: str) -> OrganisationUnitGroup:
        """Fetch one group by UID with `organisationUnits` + `groupSets` populated."""
        return await self._client.get(
            f"/api/organisationUnitGroups/{uid}", model=OrganisationUnitGroup, params={"fields": _OU_GROUP_FIELDS}
        )

    async def list_members(
        self,
        uid: str,
        *,
        page: int = 1,
        page_size: int = 50,
    ) -> list[OrganisationUnit]:
        """Page through OUs belonging to one group.

        Hits `/api/organisationUnits?filter=organisationUnitGroups.id:eq:<uid>`
        so pagination stays server-side — needed for province-level
        groups in large countries where a single group can carry
        thousands of facilities.
        """
        return cast(
            list[OrganisationUnit],
            await self._client.resources.organisation_units.list(
                fields=_MEMBER_FIELDS,
                filters=[f"organisationUnitGroups.id:eq:{uid}"],
                order=["name:asc"],
                page=page,
                page_size=page_size,
            ),
        )

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

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

    async def add_members(self, uid: str, *, ou_uids: list[str]) -> OrganisationUnitGroup:
        """Add `ou_uids` to the group without clobbering existing members.

        DHIS2 exposes a direct `POST
        /api/organisationUnitGroups/{uid}/organisationUnits/{ou_uid}`
        shortcut for this — uses that per member so each add is one
        round-trip and the server never sees a payload with the full
        current membership.
        """
        for ou_uid in ou_uids:
            await self._client.resources.organisation_unit_groups.add_collection_item(uid, "organisationUnits", ou_uid)
        return await self.get(uid)

    async def remove_members(self, uid: str, *, ou_uids: list[str]) -> OrganisationUnitGroup:
        """Drop `ou_uids` from the group via the per-member DELETE shortcut."""
        for ou_uid in ou_uids:
            await self._client.resources.organisation_unit_groups.remove_collection_item(
                uid, "organisationUnits", ou_uid
            )
        return await self.get(uid)

    async def delete(self, uid: str) -> None:
        """Delete a group — members stay, only the grouping row is removed."""
        if not uid:
            raise ValueError("delete requires a non-empty uid")
        await self._client.resources.organisation_unit_groups.delete(uid)
Functions
__init__(client)

Bind to the sharing client.

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

Return every OrganisationUnitGroup with its member refs inline.

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_unit_groups.py
async def list_all(self) -> list[OrganisationUnitGroup]:
    """Return every OrganisationUnitGroup with its member refs inline."""
    return cast(
        list[OrganisationUnitGroup],
        await self._client.resources.organisation_unit_groups.list(
            fields=_OU_GROUP_FIELDS,
            paging=False,
        ),
    )
get(uid) async

Fetch one group by UID with organisationUnits + groupSets populated.

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_unit_groups.py
async def get(self, uid: str) -> OrganisationUnitGroup:
    """Fetch one group by UID with `organisationUnits` + `groupSets` populated."""
    return await self._client.get(
        f"/api/organisationUnitGroups/{uid}", model=OrganisationUnitGroup, params={"fields": _OU_GROUP_FIELDS}
    )
list_members(uid, *, page=1, page_size=50) async

Page through OUs belonging to one group.

Hits /api/organisationUnits?filter=organisationUnitGroups.id:eq:<uid> so pagination stays server-side — needed for province-level groups in large countries where a single group can carry thousands of facilities.

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_unit_groups.py
async def list_members(
    self,
    uid: str,
    *,
    page: int = 1,
    page_size: int = 50,
) -> list[OrganisationUnit]:
    """Page through OUs belonging to one group.

    Hits `/api/organisationUnits?filter=organisationUnitGroups.id:eq:<uid>`
    so pagination stays server-side — needed for province-level
    groups in large countries where a single group can carry
    thousands of facilities.
    """
    return cast(
        list[OrganisationUnit],
        await self._client.resources.organisation_units.list(
            fields=_MEMBER_FIELDS,
            filters=[f"organisationUnitGroups.id:eq:{uid}"],
            order=["name:asc"],
            page=page,
            page_size=page_size,
        ),
    )
create(*, name, short_name, uid=None, code=None, description=None, color=None, symbol=None) async

Create an empty group; add members afterwards via add_members.

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

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

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_unit_groups.py
async def update(self, group: OrganisationUnitGroup) -> OrganisationUnitGroup:
    """PUT an edited group back. `group.id` must be set."""
    if not group.id:
        raise ValueError("update requires group.id to be set")
    body = group.model_dump(by_alias=True, exclude_none=True, mode="json")
    await self._client.put_raw(f"/api/organisationUnitGroups/{group.id}", body=body)
    return await self.get(group.id)
add_members(uid, *, ou_uids) async

Add ou_uids to the group without clobbering existing members.

DHIS2 exposes a direct POST /api/organisationUnitGroups/{uid}/organisationUnits/{ou_uid} shortcut for this — uses that per member so each add is one round-trip and the server never sees a payload with the full current membership.

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_unit_groups.py
async def add_members(self, uid: str, *, ou_uids: list[str]) -> OrganisationUnitGroup:
    """Add `ou_uids` to the group without clobbering existing members.

    DHIS2 exposes a direct `POST
    /api/organisationUnitGroups/{uid}/organisationUnits/{ou_uid}`
    shortcut for this — uses that per member so each add is one
    round-trip and the server never sees a payload with the full
    current membership.
    """
    for ou_uid in ou_uids:
        await self._client.resources.organisation_unit_groups.add_collection_item(uid, "organisationUnits", ou_uid)
    return await self.get(uid)
remove_members(uid, *, ou_uids) async

Drop ou_uids from the group via the per-member DELETE shortcut.

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_unit_groups.py
async def remove_members(self, uid: str, *, ou_uids: list[str]) -> OrganisationUnitGroup:
    """Drop `ou_uids` from the group via the per-member DELETE shortcut."""
    for ou_uid in ou_uids:
        await self._client.resources.organisation_unit_groups.remove_collection_item(
            uid, "organisationUnits", ou_uid
        )
    return await self.get(uid)
delete(uid) async

Delete a group — members stay, only the grouping row is removed.

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

organisation_unit_group_sets

OrganisationUnitGroupSet authoring — Dhis2Client.organisation_unit_group_sets.

A DHIS2 OrganisationUnitGroupSet is the analytics dimension that groups OrganisationUnitGroups: "Facility Ownership" carries the public/private/NGO groups, "Facility Type" carries urban/rural, etc. Visualisations and pivot tables reference a group set by UID to get a named axis off the OU hierarchy. This accessor wraps the generated CRUD with the membership primitives authoring flows need:

  • list_groups — enumerate groups inside one set.
  • add_groups / remove_groups — set-diff membership edits via the per-item DELETE/POST shortcut DHIS2 exposes for group→set linkage.

Classes

OrganisationUnitGroupSet

Bases: BaseModel

Generated model for DHIS2 OrganisationUnitGroupSet.

DHIS2 Organisation Unit Group Set - persisted metadata (generated from /api/schemas at DHIS2 v42).

API endpoint: /api/organisationUnitGroupSets.

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

    DHIS2 Organisation Unit Group Set - persisted metadata (generated from /api/schemas at DHIS2 v42).

    API endpoint: /api/organisationUnitGroupSets.

    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
    allItems: 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.")
    compulsory: bool | None = None
    created: datetime | None = None
    createdBy: Reference | None = Field(default=None, description="Reference to User.")
    dataDimension: bool | None = None
    dataDimensionType: DataDimensionType | None = None
    description: str | None = Field(default=None, description="Length/value min=1, max=2147483647.")
    dimension: str | None = Field(default=None, description="Length/value max=2147483647.")
    dimensionItemKeywords: Any | None = Field(
        default=None, description="Reference to DimensionItemKeywords. Read-only (inverse side)."
    )
    dimensionType: DimensionType | None = None
    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).")
    filter: str | None = Field(default=None, description="Length/value max=2147483647.")
    formName: str | None = Field(default=None, description="Length/value max=2147483647.")
    href: str | None = None
    id: str | None = Field(default=None, description="Unique. Length/value min=11, max=11.")
    includeSubhierarchyInAnalytics: bool | None = None
    items: list[Any] | None = Field(
        default=None, description="Collection of DimensionalItemObject. Read-only (inverse side)."
    )
    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).")
    name: str | None = Field(default=None, description="Unique. Length/value min=1, max=230.")
    optionSet: Reference | None = Field(default=None, description="Reference to OptionSet. Read-only (inverse side).")
    organisationUnitGroups: list[Any] | None = Field(default=None, description="Collection of OrganisationUnitGroup.")
    program: Reference | None = Field(default=None, description="Reference to Program. Read-only (inverse side).")
    programStage: Reference | None = Field(
        default=None, description="Reference to ProgramStage. Read-only (inverse side)."
    )
    repetition: Any | None = Field(default=None, description="Reference to EventRepetition. 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.")
    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).")
    valueType: ValueType | None = Field(default=None, description="Read-only.")

OrganisationUnitGroupSetsAccessor

Dhis2Client.organisation_unit_group_sets — CRUD + group-membership helpers.

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

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

    async def list_all(self) -> list[OrganisationUnitGroupSet]:
        """Return every OrganisationUnitGroupSet with its groups resolved inline."""
        return cast(
            list[OrganisationUnitGroupSet],
            await self._client.resources.organisation_unit_group_sets.list(
                fields=_OU_GROUP_SET_FIELDS,
                paging=False,
            ),
        )

    async def get(self, uid: str) -> OrganisationUnitGroupSet:
        """Fetch one group set by UID with its `organisationUnitGroups` populated."""
        return await self._client.get(
            f"/api/organisationUnitGroupSets/{uid}",
            model=OrganisationUnitGroupSet,
            params={"fields": _OU_GROUP_SET_FIELDS},
        )

    async def list_groups(self, uid: str) -> list[OrganisationUnitGroup]:
        """Return the groups that belong to one group set, in definition order."""
        group_set = await self.get(uid)
        groups = group_set.organisationUnitGroups or []
        return [OrganisationUnitGroup.model_validate(g) for g in groups if isinstance(g, dict)]

    async def create(
        self,
        *,
        name: str,
        short_name: str,
        uid: str | None = None,
        code: str | None = None,
        description: str | None = None,
        compulsory: bool = False,
        data_dimension: bool = True,
        include_subhierarchy_in_analytics: bool = False,
    ) -> OrganisationUnitGroupSet:
        """Create an empty group set; wire groups into it via `add_groups`.

        `data_dimension=True` (the default) exposes the set as an axis
        in the Pivot Table + Data Visualizer apps. `compulsory=True`
        requires every OU to land in exactly one group of the set —
        DHIS2's Data Integrity checks will flag dangling OUs otherwise.
        """
        payload: dict[str, Any] = {
            "name": name,
            "shortName": short_name,
            "compulsory": compulsory,
            "dataDimension": data_dimension,
            "includeSubhierarchyInAnalytics": include_subhierarchy_in_analytics,
        }
        if uid:
            payload["id"] = uid
        if code:
            payload["code"] = code
        if description:
            payload["description"] = description
        envelope = await self._client.post("/api/organisationUnitGroupSets", payload, model=WebMessageResponse)
        created_uid = envelope.created_uid or uid
        if not created_uid:
            raise RuntimeError("organisation-unit-group-set create did not return a uid")
        return await self.get(created_uid)

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

    async def add_groups(self, uid: str, *, group_uids: list[str]) -> OrganisationUnitGroupSet:
        """Add `group_uids` to the set via the per-item POST shortcut."""
        for group_uid in group_uids:
            await self._client.resources.organisation_unit_group_sets.add_collection_item(
                uid, "organisationUnitGroups", group_uid
            )
        return await self.get(uid)

    async def remove_groups(self, uid: str, *, group_uids: list[str]) -> OrganisationUnitGroupSet:
        """Drop `group_uids` from the set via the per-item DELETE shortcut."""
        for group_uid in group_uids:
            await self._client.resources.organisation_unit_group_sets.remove_collection_item(
                uid, "organisationUnitGroups", group_uid
            )
        return await self.get(uid)

    async def delete(self, uid: str) -> None:
        """Delete a group set — groups stay, only the dimension row is removed."""
        if not uid:
            raise ValueError("delete requires a non-empty uid")
        await self._client.resources.organisation_unit_group_sets.delete(uid)
Functions
__init__(client)

Bind to the sharing client.

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

Return every OrganisationUnitGroupSet with its groups resolved inline.

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_unit_group_sets.py
async def list_all(self) -> list[OrganisationUnitGroupSet]:
    """Return every OrganisationUnitGroupSet with its groups resolved inline."""
    return cast(
        list[OrganisationUnitGroupSet],
        await self._client.resources.organisation_unit_group_sets.list(
            fields=_OU_GROUP_SET_FIELDS,
            paging=False,
        ),
    )
get(uid) async

Fetch one group set by UID with its organisationUnitGroups populated.

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_unit_group_sets.py
async def get(self, uid: str) -> OrganisationUnitGroupSet:
    """Fetch one group set by UID with its `organisationUnitGroups` populated."""
    return await self._client.get(
        f"/api/organisationUnitGroupSets/{uid}",
        model=OrganisationUnitGroupSet,
        params={"fields": _OU_GROUP_SET_FIELDS},
    )
list_groups(uid) async

Return the groups that belong to one group set, in definition order.

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_unit_group_sets.py
async def list_groups(self, uid: str) -> list[OrganisationUnitGroup]:
    """Return the groups that belong to one group set, in definition order."""
    group_set = await self.get(uid)
    groups = group_set.organisationUnitGroups or []
    return [OrganisationUnitGroup.model_validate(g) for g in groups if isinstance(g, dict)]
create(*, name, short_name, uid=None, code=None, description=None, compulsory=False, data_dimension=True, include_subhierarchy_in_analytics=False) async

Create an empty group set; wire groups into it via add_groups.

data_dimension=True (the default) exposes the set as an axis in the Pivot Table + Data Visualizer apps. compulsory=True requires every OU to land in exactly one group of the set — DHIS2's Data Integrity checks will flag dangling OUs otherwise.

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_unit_group_sets.py
async def create(
    self,
    *,
    name: str,
    short_name: str,
    uid: str | None = None,
    code: str | None = None,
    description: str | None = None,
    compulsory: bool = False,
    data_dimension: bool = True,
    include_subhierarchy_in_analytics: bool = False,
) -> OrganisationUnitGroupSet:
    """Create an empty group set; wire groups into it via `add_groups`.

    `data_dimension=True` (the default) exposes the set as an axis
    in the Pivot Table + Data Visualizer apps. `compulsory=True`
    requires every OU to land in exactly one group of the set —
    DHIS2's Data Integrity checks will flag dangling OUs otherwise.
    """
    payload: dict[str, Any] = {
        "name": name,
        "shortName": short_name,
        "compulsory": compulsory,
        "dataDimension": data_dimension,
        "includeSubhierarchyInAnalytics": include_subhierarchy_in_analytics,
    }
    if uid:
        payload["id"] = uid
    if code:
        payload["code"] = code
    if description:
        payload["description"] = description
    envelope = await self._client.post("/api/organisationUnitGroupSets", payload, model=WebMessageResponse)
    created_uid = envelope.created_uid or uid
    if not created_uid:
        raise RuntimeError("organisation-unit-group-set create did not return a uid")
    return await self.get(created_uid)
update(group_set) async

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

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_unit_group_sets.py
async def update(self, group_set: OrganisationUnitGroupSet) -> OrganisationUnitGroupSet:
    """PUT an edited group set back. `group_set.id` must be set."""
    if not group_set.id:
        raise ValueError("update requires group_set.id to be set")
    body = group_set.model_dump(by_alias=True, exclude_none=True, mode="json")
    await self._client.put_raw(f"/api/organisationUnitGroupSets/{group_set.id}", body=body)
    return await self.get(group_set.id)
add_groups(uid, *, group_uids) async

Add group_uids to the set via the per-item POST shortcut.

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_unit_group_sets.py
async def add_groups(self, uid: str, *, group_uids: list[str]) -> OrganisationUnitGroupSet:
    """Add `group_uids` to the set via the per-item POST shortcut."""
    for group_uid in group_uids:
        await self._client.resources.organisation_unit_group_sets.add_collection_item(
            uid, "organisationUnitGroups", group_uid
        )
    return await self.get(uid)
remove_groups(uid, *, group_uids) async

Drop group_uids from the set via the per-item DELETE shortcut.

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_unit_group_sets.py
async def remove_groups(self, uid: str, *, group_uids: list[str]) -> OrganisationUnitGroupSet:
    """Drop `group_uids` from the set via the per-item DELETE shortcut."""
    for group_uid in group_uids:
        await self._client.resources.organisation_unit_group_sets.remove_collection_item(
            uid, "organisationUnitGroups", group_uid
        )
    return await self.get(uid)
delete(uid) async

Delete a group set — groups stay, only the dimension row is removed.

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_unit_group_sets.py
async def delete(self, uid: str) -> None:
    """Delete a group set — groups stay, only the dimension row is removed."""
    if not uid:
        raise ValueError("delete requires a non-empty uid")
    await self._client.resources.organisation_unit_group_sets.delete(uid)

organisation_unit_levels

OrganisationUnitLevel naming + listing — Dhis2Client.organisation_unit_levels.

DHIS2 auto-creates one OrganisationUnitLevel row per depth in the OU tree (level 1 = roots, level 2 = their children, etc.) but leaves them unnamed until an admin supplies human labels — "Country", "Province", "District", "Facility". Those labels are what the Maintenance app + analytics UIs surface in dropdowns, so a freshly seeded instance with anonymous levels is hard to navigate.

Renaming is the only common write operation. This accessor covers list / get / rename + the offline-hierarchy tuning knob (offline_levels) instances sometimes use for mobile caching. Full CRUD (create / delete) is available on the generated accessor (client.resources.organisation_unit_levels) for admins who need it.

Classes

OrganisationUnitLevel

Bases: BaseModel

Generated model for DHIS2 OrganisationUnitLevel.

DHIS2 Organisation Unit Level - persisted metadata (generated from /api/schemas at DHIS2 v42).

API endpoint: /api/organisationUnitLevels.

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

    DHIS2 Organisation Unit Level - persisted metadata (generated from /api/schemas at DHIS2 v42).

    API endpoint: /api/organisationUnitLevels.

    Field `Field(description=...)` entries flag DHIS2 semantics the bare
    type can't capture: which side of a relationship owns the link
    (writable) vs the inverse side (ignored by the API), uniqueness
    constraints, and length bounds.
    """

    model_config = ConfigDict(extra="allow", populate_by_name=True)

    access: Any | None = Field(default=None, description="Reference to Access. Read-only (inverse side).")
    attributeValues: Any | None = Field(
        default=None, description="Reference to AttributeValues. Read-only (inverse side)."
    )
    code: str | None = Field(default=None, description="Unique. Length/value max=50.")
    created: datetime | None = None
    createdBy: Reference | None = Field(default=None, description="Reference to User. Read-only (inverse side).")
    displayName: str | None = Field(default=None, description="Read-only.")
    favorite: bool | None = Field(default=None, description="Read-only.")
    favorites: list[Any] | None = Field(default=None, description="Collection of String. Read-only (inverse side).")
    href: str | None = None
    id: str | None = Field(default=None, description="Unique. Length/value min=11, max=11.")
    lastUpdated: datetime | None = None
    lastUpdatedBy: Reference | None = Field(default=None, description="Reference to User.")
    level: int | None = Field(default=None, description="Unique. Length/value min=1, max=999.")
    name: str | None = Field(default=None, description="Length/value min=1, max=230.")
    offlineLevels: int | None = Field(default=None, description="Length/value max=2147483647.")
    sharing: Any | None = Field(default=None, description="Reference to Sharing. Read-only (inverse side).")
    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).")

OrganisationUnitLevelsAccessor

Dhis2Client.organisation_unit_levels — list / get / rename OU levels.

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_unit_levels.py
class OrganisationUnitLevelsAccessor:
    """`Dhis2Client.organisation_unit_levels` — list / get / rename OU levels."""

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

    async def list_all(self) -> list[OrganisationUnitLevel]:
        """Return every level sorted by `level` ascending (roots first)."""
        raw = await self._client.get_raw(
            "/api/organisationUnitLevels",
            params={
                "fields": _OU_LEVEL_FIELDS,
                "order": "level:asc",
                "paging": "false",
            },
        )
        rows = raw.get("organisationUnitLevels") or []
        return [OrganisationUnitLevel.model_validate(row) for row in rows if isinstance(row, dict)]

    async def list_with_gaps(self) -> list[OrganisationUnitLevel]:
        """Return existing level rows + synthetic placeholders for every OU depth without a row.

        DHIS2 only persists an `OrganisationUnitLevel` row when one is
        explicitly created (via the Maintenance app or this accessor's
        `rename` methods). A freshly-seeded instance typically has rows
        only for the depths a fixture touched, so `list_all()` hides
        the full shape of the tree.

        `list_with_gaps()` fills the missing slots with synthesised
        `OrganisationUnitLevel(level=N, name=None, id=None)` entries so
        CLI/MCP renderers can show `(unnamed)` for every depth the tree
        actually has. Synthetic rows have no `id` — callers that want
        to name one should POST a new row via `/api/organisationUnitLevels`
        (or use the `rename_by_level` method which PUTs to an existing
        row only).
        """
        existing = await self.list_all()
        named_depths = {row.level for row in existing if row.level is not None}
        max_depth = await self._max_tree_depth()
        synthetic: list[OrganisationUnitLevel] = []
        for depth in range(1, max_depth + 1):
            if depth not in named_depths:
                synthetic.append(OrganisationUnitLevel(level=depth, name=None, id=None))
        merged = existing + synthetic
        merged.sort(key=lambda row: row.level or 0)
        return merged

    async def _max_tree_depth(self) -> int:
        """Cheapest way to ask "how deep does the OU tree go?" — one paged GET.

        Hits `/api/organisationUnits?fields=level&order=level:desc&pageSize=1`
        so the server returns exactly one OU — the deepest one. Zero if
        the instance has no OUs (unusual but handled).
        """
        raw = await self._client.get_raw(
            "/api/organisationUnits",
            params={"fields": "level", "order": "level:desc", "pageSize": "1"},
        )
        rows = raw.get("organisationUnits") or []
        if not rows or not isinstance(rows[0], dict):
            return 0
        level = rows[0].get("level")
        return int(level) if isinstance(level, int | str) and str(level).isdigit() else 0

    async def get(self, uid: str) -> OrganisationUnitLevel:
        """Fetch one level row by UID."""
        return await self._client.get(
            f"/api/organisationUnitLevels/{uid}", model=OrganisationUnitLevel, params={"fields": _OU_LEVEL_FIELDS}
        )

    async def get_by_level(self, level: int) -> OrganisationUnitLevel | None:
        """Resolve a level row by its numeric depth (1 = roots)."""
        raw = await self._client.get_raw(
            "/api/organisationUnitLevels",
            params={
                "fields": _OU_LEVEL_FIELDS,
                "filter": f"level:eq:{level}",
                "paging": "false",
            },
        )
        rows = raw.get("organisationUnitLevels") or []
        for row in rows:
            if isinstance(row, dict):
                return OrganisationUnitLevel.model_validate(row)
        return None

    async def rename(
        self,
        uid: str,
        *,
        name: str,
        code: str | None = None,
        offline_levels: int | None = None,
    ) -> OrganisationUnitLevel:
        """Rename a level (and optionally tweak `code` / `offlineLevels`).

        DHIS2 rejects PATCH on this endpoint, so the shape is "GET,
        mutate, PUT". Returns the freshly-fetched level so the caller
        sees the persisted row.
        """
        level = await self.get(uid)
        body = level.model_dump(by_alias=True, exclude_none=True, mode="json")
        body["name"] = name
        if code is not None:
            body["code"] = code
        if offline_levels is not None:
            body["offlineLevels"] = offline_levels
        await self._client.put_raw(f"/api/organisationUnitLevels/{uid}", body=body)
        return await self.get(uid)

    async def rename_by_level(
        self,
        level: int,
        *,
        name: str,
        code: str | None = None,
        offline_levels: int | None = None,
    ) -> OrganisationUnitLevel:
        """Upsert the label for a numeric depth — PUT if a row exists, POST otherwise.

        DHIS2 only persists a level row when one is explicitly created,
        so "rename level 2 to 'Province'" on a fresh instance means
        creating the row, not updating it. This method handles both
        cases transparently so callers don't need to know whether the
        row existed.
        """
        existing = await self.get_by_level(level)
        if existing is not None and existing.id:
            return await self.rename(existing.id, name=name, code=code, offline_levels=offline_levels)
        payload: dict[str, Any] = {"level": level, "name": name}
        if code is not None:
            payload["code"] = code
        if offline_levels is not None:
            payload["offlineLevels"] = offline_levels
        envelope = await self._client.post("/api/organisationUnitLevels", payload, model=WebMessageResponse)
        created_uid = envelope.created_uid
        if created_uid:
            return await self.get(created_uid)
        refetched = await self.get_by_level(level)
        if refetched is None:
            raise RuntimeError(f"failed to create OrganisationUnitLevel at depth {level}")
        return refetched
Functions
__init__(client)

Bind to the sharing client.

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

Return every level sorted by level ascending (roots first).

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_unit_levels.py
async def list_all(self) -> list[OrganisationUnitLevel]:
    """Return every level sorted by `level` ascending (roots first)."""
    raw = await self._client.get_raw(
        "/api/organisationUnitLevels",
        params={
            "fields": _OU_LEVEL_FIELDS,
            "order": "level:asc",
            "paging": "false",
        },
    )
    rows = raw.get("organisationUnitLevels") or []
    return [OrganisationUnitLevel.model_validate(row) for row in rows if isinstance(row, dict)]
list_with_gaps() async

Return existing level rows + synthetic placeholders for every OU depth without a row.

DHIS2 only persists an OrganisationUnitLevel row when one is explicitly created (via the Maintenance app or this accessor's rename methods). A freshly-seeded instance typically has rows only for the depths a fixture touched, so list_all() hides the full shape of the tree.

list_with_gaps() fills the missing slots with synthesised OrganisationUnitLevel(level=N, name=None, id=None) entries so CLI/MCP renderers can show (unnamed) for every depth the tree actually has. Synthetic rows have no id — callers that want to name one should POST a new row via /api/organisationUnitLevels (or use the rename_by_level method which PUTs to an existing row only).

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_unit_levels.py
async def list_with_gaps(self) -> list[OrganisationUnitLevel]:
    """Return existing level rows + synthetic placeholders for every OU depth without a row.

    DHIS2 only persists an `OrganisationUnitLevel` row when one is
    explicitly created (via the Maintenance app or this accessor's
    `rename` methods). A freshly-seeded instance typically has rows
    only for the depths a fixture touched, so `list_all()` hides
    the full shape of the tree.

    `list_with_gaps()` fills the missing slots with synthesised
    `OrganisationUnitLevel(level=N, name=None, id=None)` entries so
    CLI/MCP renderers can show `(unnamed)` for every depth the tree
    actually has. Synthetic rows have no `id` — callers that want
    to name one should POST a new row via `/api/organisationUnitLevels`
    (or use the `rename_by_level` method which PUTs to an existing
    row only).
    """
    existing = await self.list_all()
    named_depths = {row.level for row in existing if row.level is not None}
    max_depth = await self._max_tree_depth()
    synthetic: list[OrganisationUnitLevel] = []
    for depth in range(1, max_depth + 1):
        if depth not in named_depths:
            synthetic.append(OrganisationUnitLevel(level=depth, name=None, id=None))
    merged = existing + synthetic
    merged.sort(key=lambda row: row.level or 0)
    return merged
get(uid) async

Fetch one level row by UID.

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_unit_levels.py
async def get(self, uid: str) -> OrganisationUnitLevel:
    """Fetch one level row by UID."""
    return await self._client.get(
        f"/api/organisationUnitLevels/{uid}", model=OrganisationUnitLevel, params={"fields": _OU_LEVEL_FIELDS}
    )
get_by_level(level) async

Resolve a level row by its numeric depth (1 = roots).

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_unit_levels.py
async def get_by_level(self, level: int) -> OrganisationUnitLevel | None:
    """Resolve a level row by its numeric depth (1 = roots)."""
    raw = await self._client.get_raw(
        "/api/organisationUnitLevels",
        params={
            "fields": _OU_LEVEL_FIELDS,
            "filter": f"level:eq:{level}",
            "paging": "false",
        },
    )
    rows = raw.get("organisationUnitLevels") or []
    for row in rows:
        if isinstance(row, dict):
            return OrganisationUnitLevel.model_validate(row)
    return None
rename(uid, *, name, code=None, offline_levels=None) async

Rename a level (and optionally tweak code / offlineLevels).

DHIS2 rejects PATCH on this endpoint, so the shape is "GET, mutate, PUT". Returns the freshly-fetched level so the caller sees the persisted row.

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_unit_levels.py
async def rename(
    self,
    uid: str,
    *,
    name: str,
    code: str | None = None,
    offline_levels: int | None = None,
) -> OrganisationUnitLevel:
    """Rename a level (and optionally tweak `code` / `offlineLevels`).

    DHIS2 rejects PATCH on this endpoint, so the shape is "GET,
    mutate, PUT". Returns the freshly-fetched level so the caller
    sees the persisted row.
    """
    level = await self.get(uid)
    body = level.model_dump(by_alias=True, exclude_none=True, mode="json")
    body["name"] = name
    if code is not None:
        body["code"] = code
    if offline_levels is not None:
        body["offlineLevels"] = offline_levels
    await self._client.put_raw(f"/api/organisationUnitLevels/{uid}", body=body)
    return await self.get(uid)
rename_by_level(level, *, name, code=None, offline_levels=None) async

Upsert the label for a numeric depth — PUT if a row exists, POST otherwise.

DHIS2 only persists a level row when one is explicitly created, so "rename level 2 to 'Province'" on a fresh instance means creating the row, not updating it. This method handles both cases transparently so callers don't need to know whether the row existed.

Source code in packages/dhis2w-client/src/dhis2w_client/organisation_unit_levels.py
async def rename_by_level(
    self,
    level: int,
    *,
    name: str,
    code: str | None = None,
    offline_levels: int | None = None,
) -> OrganisationUnitLevel:
    """Upsert the label for a numeric depth — PUT if a row exists, POST otherwise.

    DHIS2 only persists a level row when one is explicitly created,
    so "rename level 2 to 'Province'" on a fresh instance means
    creating the row, not updating it. This method handles both
    cases transparently so callers don't need to know whether the
    row existed.
    """
    existing = await self.get_by_level(level)
    if existing is not None and existing.id:
        return await self.rename(existing.id, name=name, code=code, offline_levels=offline_levels)
    payload: dict[str, Any] = {"level": level, "name": name}
    if code is not None:
        payload["code"] = code
    if offline_levels is not None:
        payload["offlineLevels"] = offline_levels
    envelope = await self._client.post("/api/organisationUnitLevels", payload, model=WebMessageResponse)
    created_uid = envelope.created_uid
    if created_uid:
        return await self.get(created_uid)
    refetched = await self.get_by_level(level)
    if refetched is None:
        raise RuntimeError(f"failed to create OrganisationUnitLevel at depth {level}")
    return refetched