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
|