client Module#

DHIS2 REST API client for Episia.

This module provides the DHIS2Client class for connecting to DHIS2 instances, fetching surveillance data, and converting it to Episia’s SurveillanceDataset format.

Class#

class episia.dhis2.client.DHIS2Client(url, username, password, api_token=None, timeout=30, verify=True)[source]#

Bases: object

DHIS2 REST API client for Episia.

Connects to any DHIS2 instance and converts data to SurveillanceDataset for immediate analysis with Episia.

Requires requests: pip install episia[dhis2]

Parameters:

url (str) – Base URL of the DHIS2 instance (no trailing slash).
username (str) – DHIS2 username.
password (str) – DHIS2 password.
timeout (int) – HTTP timeout in seconds (default 30).
verify (bool) – SSL certificate verification (default True).
api_token (Optional[str])

Example:

from episia.dhis2 import DHIS2Client

client = DHIS2Client(
    url      = "https://play.dhis2.org/40.2.2",
    username = "admin",
    password = "district",
)

# Check connection
client.ping()

# Fetch data as SurveillanceDataset
# Use semicolons for multiple periods
ds = client.to_dataset(
    data_element = "FTRrcoaog83",
    period       = "2024W01;2024W02;2024W03",
    org_unit     = "ImspTQPwCqd",
)
print(ds)
ds.to_timeseries_result().plot().show()

__init__(url, username, password, api_token=None, timeout=30, verify=True)[source]#

Parameters:

url (str) – Base URL of the DHIS2 instance.
username (str) – DHIS2 username (used when api_token is None).
password (str) – DHIS2 password (used when api_token is None).
api_token (str | None) – Personal access token (PAT). When provided, Basic Auth is not used. Generate one at: Profile -> Personal access tokens in DHIS2.
timeout (int) – HTTP timeout in seconds (default 30).
verify (bool) – SSL certificate verification (default True).

__repr__()[source]#

Return repr(self).

Return type:: str

fetch_analytics(data_elements, period, org_unit, org_unit_mode='SELECTED')[source]#

Fetch data from the DHIS2 /api/analytics endpoint.

Parameters:

data_elements (str | List[str]) – Data element UID or list of UIDs.
period (str) –
Period expression. Examples:

Single period:
```
'202401'
'2024W01'
```
Multiple periods (semicolon-separated):
```
'202401;202402;202403'
'2024W01;2024W02;2024W03'
```
Relative periods:
```
'LAST_12_WEEKS'
'LAST_6_MONTHS'
'THIS_YEAR'
```
Warning

The colon range syntax (e.g. '2024W01:2024W52') is NOT supported by the DHIS2 analytics API for monthly or quarterly periods. Always use semicolons to list periods explicitly, or use a relative period keyword.
org_unit (str) – Organisation unit UID.
org_unit_mode (str) – SELECTED, CHILDREN, or DESCENDANTS.

Returns:

Raw DHIS2 analytics JSON response.

Return type:

Dict[str, Any]

fetch_data_value_sets(data_set, period, org_unit)[source]#

Fetch raw data values from /api/dataValueSets.

Parameters:

data_set (str) – Data set UID.
period (str) – Period string (e.g. ‘202401’).
org_unit (str) – Organisation unit UID.

Returns:

Raw DHIS2 dataValueSets JSON response.

Return type:

Dict[str, Any]

list_data_elements(fields='id,name,shortName,valueType')[source]#

List available data elements from DHIS2.

Parameters:: fields (str) – Comma-separated fields to return.
Returns:: List of data element dicts.
Return type:: List[Dict[str, Any]]

list_org_units(level=None, parent=None, fields='id,name,level,shortName')[source]#

List organisation units from DHIS2.

Parameters:

level (int | None) – Filter by hierarchy level (1=national, 2=region…).
parent (str | None) – Filter by parent UID.
fields (str) – Comma-separated fields to return.

Returns:

List of organisation unit dicts.

Return type:

List[Dict[str, Any]]

ping()[source]#

Test connection to the DHIS2 instance.

Returns:: True if connection is successful.
Raises:: ConnectionError – if the server is unreachable or credentials invalid.
Return type:: bool

to_dataset(data_element, period, org_unit, deaths_element=None, org_unit_mode='SELECTED')[source]#

Fetch data from DHIS2 and return a ready-to-use SurveillanceDataset.

This is the main entry point for Episia-DHIS2 integration.

Parameters:

data_element (str) – UID of the cases data element.
period (str) – Period expression. Use semicolons for multiple periods (e.g. '202401;202402;202403') or a relative keyword (e.g. 'LAST_12_MONTHS').
org_unit (str) – Organisation unit UID.
deaths_element (str | None) – UID of deaths data element (optional).
org_unit_mode (str) – SELECTED, CHILDREN, or DESCENDANTS.

Returns:

SurveillanceDataset ready for Episia analysis.

Return type:

SurveillanceDataset

Example:

ds = client.to_dataset(
    data_element = "FTRrcoaog83",
    period       = "LAST_52_WEEKS",
    org_unit     = "ImspTQPwCqd",
)
engine = AlertEngine(ds)
alerts = engine.run(threshold=15)

to_dataset_by_district(data_element, period, org_unit)[source]#

Fetch data disaggregated by child org units (districts).

Equivalent to to_dataset() with org_unit_mode=’CHILDREN’.

Returns:

SurveillanceDataset with district_col=’org_unit’.

Parameters:

data_element (str)
period (str)
org_unit (str)

Return type:

SurveillanceDataset

Examples#

Basic connection and data fetching:

from episia.dhis2 import DHIS2Client

# Connect to DHIS2 demo instance
client = DHIS2Client(
    url      = "https://play.dhis2.org/40.2.2",
    username = "admin",
    password = "district",
)

# Test connection
if client.ping():
    print("Connected successfully!")

# Fetch data as SurveillanceDataset
ds = client.to_dataset(
    data_element = "FTRrcoaog83",  # Malaria cases
    period       = "LAST_52_WEEKS",
    org_unit     = "ImspTQPwCqd",   # Sierra Leone
)

print(f"Total cases: {ds.total_cases}")
ds.to_timeseries_result().plot().show()

Fetching with deaths data:

ds = client.to_dataset(
    data_element   = "FTRrcoaog83",   # Cases
    deaths_element = "cYeuwXTCPkU",   # Deaths
    period         = "2024W01:2024W52",
    org_unit       = "ImspTQPwCqd",
)

print(f"CFR: {ds.cfr:.1%}")

District-level data:

# Fetch data for all districts under the national level
ds = client.to_dataset_by_district(
    data_element = "FTRrcoaog83",
    period       = "2024W01:2024W52",
    org_unit     = "ImspTQPwCqd",   # National level
)

# Now dataset has district column
for district in ds.districts:
    district_ds = ds.filter_district(district)
    print(f"{district}: {district_ds.total_cases} cases")

Advanced usage with raw API calls:

# List organisation units
org_units = client.list_org_units(level=2)  # Regions
for ou in org_units:
    print(f"{ou['name']}: {ou['id']}")

# List data elements
elements = client.list_data_elements()
for elem in elements[:5]:  # First 5
    print(f"{elem['name']}: {elem['id']}")

# Raw analytics fetch
raw_data = client.fetch_analytics(
    data_elements = ["FTRrcoaog83", "cYeuwXTCPkU"],
    period        = "2024W01:2024W52",
    org_unit      = "ImspTQPwCqd",
)