Simple, synchronous DHIS2 Web API client — dict/JSON only, Jupyter-friendly (no context managers), clean paging, read-only users, CRUD for metadata (org units, data elements, data sets), data values & data value sets, analytics, and minimal logging by default.
- Why this client?
- Installation
- Requirements
- Quickstart
- Authentication & Settings
- Logging
- Paging
- Collections: page-by-page iteration
- Convenience Methods
- Examples
- Raw API calls
- Testing
- Dev Setup
- Integration Tests
- Roadmap
- License
DHIS2 is one of the most widely used health information platforms worldwide, and Python is a popular choice in data science, research, and integration workflows. Bringing the two together makes it easier to build analytics, integrations, and automation around DHIS2 data.
This client provides a lightweight and simple way to work with the DHIS2 Web API:
- Always returns plain Python
dict
/ JSON objects — no custom models or ORM layers. - Handles paging cleanly, so you can iterate over large DHIS2 collections without surprises.
- Offers convenience methods for common entities (users, organisation units, data elements, data sets, data values, analytics) while keeping full access to the raw API.
- Keeps setup minimal — synchronous, Jupyter-friendly, and easy to configure.
Requires Python 3.10+
pip install dhis2-client
# or, from source
pip install -e .
- Python 3.10+
- DHIS2 server URL and valid credentials (Basic or token)
from dhis2_client import DHIS2Client
client = DHIS2Client(
base_url="http://localhost:8080",
username="admin",
password="district", # Basic auth by default
)
# Iterate users (read-only)
for u in client.get_users(fields="id,username", order="username:asc"):
print(u)
# Fetch all data elements into a list (respecting paging)
all_des = client.fetch_all("/api/dataElements", params={"fields": "id,displayName"})
You can configure the client directly with kwargs or centrally with a ClientSettings object.
from dhis2_client import DHIS2Client
from dhis2_client.settings import ClientSettings
# Recommended: central settings
cfg = ClientSettings(
base_url="http://localhost:8080",
username="admin",
password="district",
log_level="INFO", # default "WARNING"
log_format="json", # default "json"; use "text" for human-readable
log_destination="stdout" # default "stderr"; can also be file path
)
client = DHIS2Client(settings=cfg)
info = client.get_system_info()
print(info["version"])
Kwargs override settings if both are provided:
client = DHIS2Client(settings=cfg, log_level="DEBUG") # DEBUG takes precedence
- Default: JSON logs at WARNING level to stderr.
- Configurable: via ClientSettings or constructor kwargs.
# JSON (default) logs at INFO to stdout
cfg = ClientSettings(
base_url="http://localhost:8080",
username="admin",
password="district",
log_level="INFO", log_destination="stdout")
client = DHIS2Client(settings=cfg)
# Human-readable text logs
client = DHIS2Client(
"http://localhost:8080",
username="admin",
password="district",
log_level="INFO", log_format="text")
# File logging
client = DHIS2Client("http://localhost:8080",
username="admin", password="district",
log_level="DEBUG", log_destination="/tmp/dhis2_client.log")
Example output:
{"ts":"2025-09-19T14:20:01+0000","level":"INFO","logger":"dhis2_client","message":"Request GET /api/system/info params=None"}
- Default
pageSize=50
. get_*s()
yield items across pages.fetch_all()
returns a list of all items.
for ou in client.get_organisation_units(level=2, fields="id,displayName"):
...
All collection convenience methods (get_data_elements
, get_users
, get_organisation_units
, get_data_sets
, …) fetch results page by page from DHIS2 until all matching items are returned.
- ✅ Safe for large DHIS2 servers (does not load everything in one huge response).
- ✅ Transparent pass-through: you control
page
,pageSize
,filter
,fields
, etc. - ❌ These methods do not include the
pager
block that DHIS2 returns. Useclient.get(...)
directly if you need that metadata.
Iterate over all matching data elements (fetches pages of 50 by default):
for de in client.get_data_elements(fields="id,displayName"):
print(de["id"], de["displayName"])
Materialize in memory (not recommended for huge datasets):
des = list(client.get_data_elements(fields="id,displayName"))
print(len(des)) # total number of matching items across all pages
Get paging info (total, page count, etc.) directly from DHIS2:
raw = client.get("/api/dataElements", params={"page": 1, "pageSize": 50})
print(raw["pager"]["total"])
get(path, params=None) -> dict
post(path, json=None) -> dict
put(path, json=None) -> dict
delete(path, params=None) -> dict
list_paged(path, params=None, page_size=None, item_key=None) -> Iterable[dict]
fetch_all(path, params=None, item_key=None) -> list[dict]
get_system_info() -> dict
get_users(**filters) -> Iterable[dict]
get_user(uid, *, fields=None) -> dict
get_organisation_units(**filters) -> Iterable[dict]
get_org_unit(uid, *, fields=None) -> dict
create_org_unit(payload) -> dict
update_org_unit(uid, payload) -> dict
delete_org_unit(uid) -> dict
get_org_unit_tree(root_uid=None, levels=None) -> dict
get_organisation_units_geojson(**params) -> dict
get_org_unit_geojson(uid, **params) -> dict
get_data_elements(**filters) -> Iterable[dict]
get_data_element(uid, *, fields=None) -> dict
create_data_element(payload) -> dict
update_data_element(uid, payload) -> dict
delete_data_element(uid) -> dict
get_data_sets(**filters) -> Iterable[dict]
get_data_set(uid, *, fields=None) -> dict
create_data_set(payload) -> dict
update_data_set(uid, payload) -> dict
delete_data_set(uid) -> dict
get_data_value(de, pe, ou, co=None, aoc=None, cc=None, cp=None) -> dict
set_data_value(de, pe, ou, value, **kwargs) -> dict
delete_data_value(de, pe, ou, **kwargs) -> dict
get_data_value_set(params: dict) -> dict
post_data_value_set(payload: dict) -> dict
get_analytics(table: str = "analytics", **params) -> dict
# List users
for u in client.get_users(fields="id,username", order="username:asc"):
print(u)
# Single user
user = client.get_user("u123", fields="id,username,displayName")
# Iterate OU level 2
for ou in client.get_organisation_units(level=2, fields="id,displayName"):
print(ou)
# Single OU
ou = client.get_org_unit("ou123", fields="id,displayName")
# Create/Update/Delete OU
client.create_org_unit({"name": "Clinic A", "shortName": "ClinicA", "openingDate": "2020-01-01"})
client.update_org_unit("ou123", {"name": "Clinic Alpha"})
client.delete_org_unit("ou123")
# Tree
tree = client.get_org_unit_tree(root_uid="ouROOT")
# Collection as GeoJSON (unpaged FeatureCollection)
fc = client.get_organisation_units_geojson(level=2, fields="id,displayName,geometry")
# Single org unit as GeoJSON
feat = client.get_org_unit_geojson("ou123", fields="id,displayName,geometry")
# List
for de in client.get_data_elements(fields="id,displayName", filter=["valueType:eq:INTEGER"]):
print(de)
# CRUD
client.create_data_element({"name": "New DE", "shortName": "NDE", "valueType": "NUMBER"})
de = client.get_data_element("de123", fields="id,displayName,valueType")
client.update_data_element("de123", {"valueType": "INTEGER"})
client.delete_data_element("de123")
for ds in client.get_data_sets(fields="id,displayName"):
print(ds)
ds = client.get_data_set("ds123", fields="id,displayName")
client.create_data_set({"name": "My DS", "periodType": "Monthly"})
client.update_data_set("ds123", {"name": "My DS (Updated)"})
client.delete_data_set("ds123")
# Single data value lifecycle
client.set_data_value(de="de1", pe="202401", ou="ou1", value="42")
val = client.get_data_value(de="de1", pe="202401", ou="ou1")
client.delete_data_value(de="de1", pe="202401", ou="ou1")
# Pull a batch
dvs = client.get_data_value_set({"dataSet": "ds1", "period": "202401", "orgUnit": "ou1"})
# Push a batch
client.post_data_value_set({
"dataSet": "ds1",
"orgUnit": "ou1",
"period": "202401",
"dataValues": [
{"dataElement": "de1", "categoryOptionCombo": "co1", "value": "5"},
{"dataElement": "de2", "categoryOptionCombo": "co1", "value": "9"}
]
})
pivot = client.get_analytics(
dimension=["dx:de1;de2", "pe:LAST_12_MONTHS", "ou:LEVEL-2"],
displayProperty="NAME",
skipMeta=True,
)
Not every DHIS2 endpoint has a convenience wrapper yet. You can always use the core methods to call any path directly:
# Arbitrary GET
resp = client.get("/api/indicators", params={"fields": "id,displayName"})
# Single item
indicator = client.get("/api/indicators/abc123", params={"fields": "id,displayName"})
# Create
ou = client.post("/api/organisationUnits", json={
"name": "Clinic A",
"shortName": "ClinicA",
"openingDate": "2020-01-01"
})
# Update
client.put("/api/dataElements/de123", json={"valueType": "INTEGER"})
# Delete
client.delete("/api/dataSets/ds123")
# Iterate paged collection
for de in client.list_paged(
"/api/dataElements",
params={"fields": "id,displayName"},
item_key="dataElements"
):
print(de)
- Run unit tests (mocked; no .env needed):
pytest -q
- Lint/format:
ruff check .
ruff format .
pip install -r requirements-dev.txt
Includes: pytest
, ruff
, respx
, python-dotenv
.
Read-only integration tests (if you have credentials):
export DHIS2_BASE_URL="http://localhost:8080"
export DHIS2_USERNAME="admin"
export DHIS2_PASSWORD="district"
pytest -m integration -q
Destructive tests (opt-in; be careful):
export DHIS2_ALLOW_MUTATIONS=true
pytest -m integration -q tests/integration/test_live_mutations.py
- Stabilize core API and paging
- Optional async & CLI (later)
- More helpers (e.g., file resources, indicators)
BSD-3-Clause