A Python library for DHIS2 wrapping requests for general-purpose API interaction with DHIS2. It attempts to be useful for any data/metadata import and export tasks including various utilities like file loading, UID generation and logging. A strong focus is on JSON.
Supported and tested on Linux/macOS, Windows and DHIS2 versions >= 2.25. Python 3.6+ is required.
Contents
Python 3.6+ is required.
pip install dhis2.pyFor instructions on installing Python / pip for your operating system see realpython.com/installing-python.
Note: this project is not related with the module dhis2 which is installed with pip install dhis2. However, the import statement is for example from dhis2 import Api which is similar to the other dhis2 module.
Create an Api object:
from dhis2 import Api
api = Api('play.dhis2.org/demo', 'admin', 'district')Then run requests on it:
r = api.get('organisationUnits/Rp268JB6Ne4', params={'fields': 'id,name'})
print(r.json())
# { "name": "Adonkia CHP", "id": "Rp268JB6Ne4" }
r = api.post('metadata', json={'dataElements': [ ... ] })
print(r.status_code) # 200api.get()api.post()api.put()api.patch()api.delete()
see below for more methods.
They all return a Response object from requests
except noted otherwise. This means methods and attributes are equally available
(e.g. Response.url, Response.text, Response.status_code etc.).
Create an API object
from dhis2 import Api
api = Api('play.dhis2.org/demo', 'admin', 'district')optional arguments:
api_version: DHIS2 API versionuser_agent: submit your own User-Agent header. This is useful if you need to parse e.g. Nginx logs later.
Load from a auth JSON file in order to not store credentials in scripts. Must have the following structure:
{
"dhis": {
"baseurl": "http:https://localhost:8080",
"username": "admin",
"password": "district"
}
}from dhis2 import Api
api = Api.from_auth_file('path/to/auth.json', api_version=29, user_agent='myApp/1.0')If no file path is specified, it tries to find a file called dish.json in:
- the
DHIS_HOMEenvironment variable - your Home folder
API version as a string:
print(api.version)
# '2.30'API version as an integer:
print(api.version_int)
# 30API revision / build:
print(api.revision)
# '17f7f0b'API URL:
print(api.api_url)
# 'https://play.dhis2.org/demo/api/30'Base URL:
print(api.base_url)
# 'https://play.dhis2.org/demo'system info (this is persisted across the session):
print(api.info)
# {
# "lastAnalyticsTableRuntime": "11 m, 51 s",
# "systemId": "eed3d451-4ff5-4193-b951-ffcc68954299",
# "contextPath": "https://play.dhis2.org/2.30",
# ...Normal method: api.get(), e.g.
r = api.get('organisationUnits/Rp268JB6Ne4', params={'fields': 'id,name'})
data = r.json()Parameters:
- timeout: to override the timeout value (default: 5 seconds) in order to prevent the client to wait indefinitely on a server response.
Paging for larger GET requests via api.get_paged()
Two possible ways:
- Process every page as they come in:
for page in api.get_paged('organisationUnits', page_size=100):
print(page)
# { "organisationUnits": [ {...}, {...} ] } (100 organisationUnits)- Load all pages before proceeding (this may take a long time) - to do this, do not use
forand addmerge=True:
all_pages = api.get_paged('organisationUnits', page_size=100, merge=True):
print(all_pages)
# { "organisationUnits": [ {...}, {...} ] } (all organisationUnits)Note: Returns directly a JSON object, not a requests.Response object unlike normal GETs.
Get SQL View data as if you'd open a CSV file, optimized for larger payloads, via api.get_sqlview()
# poll a sqlView of type VIEW or MATERIALIZED_VIEW:
for row in api.get_sqlview('YOaOY605rzh', execute=True, criteria={'name': '0-11m'}):
print(row)
# {'code': 'COC_358963', 'name': '0-11m'}
# similarly, poll a sqlView of type QUERY:
for row in api.get_sqlview('qMYMT0iUGkG', var={'valueType': 'INTEGER'}):
print(row)
# if you want a list directly, cast it to a ``list`` or add ``merge=True``:
data = list(api.get_sqlview('qMYMT0iUGkG', var={'valueType': 'INTEGER'}))
# OR
# data = api.get_sqlview('qMYMT0iUGkG', var={'valueType': 'INTEGER'}, merge=True)Note: Returns directly a JSON object, not a requests.response object unlike normal GETs.
Beginning of 2.26 you can also use normal filtering on sqlViews. In that case, it's recommended
to use the stream=True parameter of the Dhis.get() method.
Usually defaults to JSON but you can get other file types:
r = api.get('organisationUnits/Rp268JB6Ne4', file_type='xml')
print(r.text)
# <?xml version='1.0' encoding='UTF-8'?><organisationUnit ...
r = api.get('organisationUnits/Rp268JB6Ne4', file_type='pdf')
with open('/path/to/file.pdf', 'wb') as f:
f.write(r.content)Normal methods:
api.post()api.put()api.patch()api.delete()
If you have such a large payload (e.g. metadata imports) that you frequently get a HTTP Error:
413 Request Entity Too Large response e.g. from Nginx you might benefit from using
the following method that splits your payload in partitions / chunks and posts them one-by-one.
You define the amount of elements in each POST by specifying a number in thresh (default: 1000).
Note that it is only possible to submit one key per payload (e.g. dataElements only, not additionally organisationUnits in the same payload).
api.post_partitioned()
import json
data = {
"organisationUnits": [
{...},
{...} # very large number of org units
]
{
for response in api.post_partitioned('metadata', json=data, thresh=5000):
text = json.loads(response.text)
print('[{}] - {}'.format(text['status'], json.dumps(text['stats'])))If you need to pass multiple parameters to your request with the same key, you may submit as a list of tuples instead when e.g.:
r = api.get('dataValueSets', params=[
('dataSet', 'pBOMPrpg1QX'), ('dataSet', 'BfMAe6Itzgt'),
('orgUnit', 'YuQRtpLP10I'), ('orgUnit', 'vWbkYPRmKyS'),
('startDate', '2013-01-01'), ('endDate', '2013-01-31')
]
)alternatively:
r = api.get('dataValueSets', params={
'dataSet': ['pBOMPrpg1QX', 'BfMAe6Itzgt'],
'orgUnit': ['YuQRtpLP10I', 'vWbkYPRmKyS'],
'startDate': '2013-01-01',
'endDate': '2013-01-31'
})from dhis2 import load_json
json_data = load_json('/path/to/file.json')
print(json_data)
# { "id": ... }Via a Python generator:
from dhis2 import load_csv
for row in load_csv('/path/to/file.csv'):
print(row)
# { "id": ... }Via a normal list, loaded fully into memory:
data = list(load_csv('/path/to/file.csv'))Create a DHIS2 UID:
uid = generate_uid()
print(uid)
# 'Rp268JB6Ne4'To create a list of 1000 UIDs:
uids = [generate_uid() for _ in range(1000)]Check if something is a valid DHIS2 UID:
uid = 'MmwcGkxy876'
print(is_valid_uid(uid))
# True
uid = 25329
print(is_valid_uid(uid))
# False
uid = 'MmwcGkxy876 '
print(is_valid_uid(uid))
# FalseUseful for deep-removing certain keys in an object,
e.g. remove all sharing by recursively removing all user and userGroupAccesses fields.
from dhis2 import clean_obj
metadata = {
"dataElements": [
{
"name": "ANC 1st visit",
"id": "fbfJHSPpUQD",
"publicAccess": "rw------",
"userGroupAccesses": [
{
"access": "r-r-----",
"userGroupUid": "Rg8wusV7QYi",
"displayName": "HIV Program Coordinators",
"id": "Rg8wusV7QYi"
},
{
"access": "rwr-----",
"userGroupUid": "qMjBflJMOfB",
"displayName": "Family Planning Program",
"id": "qMjBflJMOfB"
}
]
}
],
"dataSets": [
{
"name": "ART monthly summary",
"id": "lyLU2wR22tC",
"publicAccess": "rwr-----",
"userGroupAccesses": [
{
"access": "r-rw----",
"userGroupUid": "GogLpGmkL0g",
"displayName": "_DATASET_Child Health Program Manager",
"id": "GogLpGmkL0g"
}
]
}
]
}
cleaned = clean_obj(metadata, ['userGroupAccesses', 'publicAccess'])
pretty_json(cleaned)Which would eventually recursively remove all keys matching to userGroupAccesses or publicAccess:
{
"dataElements": [
{
"name": "ANC 1st visit",
"id": "fbfJHSPpUQD"
}
],
"dataSets": [
{
"name": "ART monthly summary",
"id": "lyLU2wR22tC"
}
]
}Print easy-readable JSON objects with colors, utilizes Pygments.
from dhis2 import pretty_json
obj = {"dataElements": [{"name": "Accute Flaccid Paralysis (Deaths < 5 yrs)", "id": "FTRrcoaog83", "aggregationType": "SUM"}]}
pretty_json(obj)... prints (in a terminal it will have colors):
{
"dataElements": [
{
"aggregationType": "SUM",
"id": "FTRrcoaog83",
"name": "Accute Flaccid Paralysis (Deaths < 5 yrs)"
}
]
}Check the importSummary response from e.g. /api/dataValues or /api/metadata import. Returns true if import went well, false if there are ignored values or the status reports not a OK or SUCCESS. This can be useful if the response returns a 200 OK but the summary reports ignored data.
from dhis2 import import_response_ok
# response as e.g. from response = api.post('metadata', data=payload).json()
response = {
"description": "The import process failed: java.lang.String cannot be cast to java.lang.Boolean",
"importCount": {
"deleted": 0,
"ignored": 1,
"imported": 0,
"updated": 0
},
"responseType": "ImportSummary",
"status": "WARNING"
}
import_successful = import_response_ok(response)
print(import_successful)
# FalseLogging utilizes logzero.
- Color output depending on log level
- DHIS2 log format including the line of the caller
- optional
logfile=specifies a rotating log file path (20 x 10MB files)
from dhis2 import setup_logger, logger
setup_logger(logfile='/var/log/app.log')
logger.info('my log message')
logger.warning('missing something')
logger.error('something went wrong')
logger.exception('with stacktrace')* INFO 2018-06-01 18:19:40,001 my log message [script:86] * ERROR 2018-06-01 18:19:40,007 something went wrong [script:87]
Use setup_logger(include_caller=False) if you want to remove [script:86] from logs.
There are two exceptions:
RequestException: DHIS2 didn't like what you requested. See the exception'scode,urlanddescription.ClientException: Something didn't work with the client not involving DHIS2.
They both inherit from Dhis2PyException.
- Real-world script examples can be found in the
examplesfolder. - dhis2.py is used in dhis2-pk (dhis2-pocket-knife)
Versions changelog
Feedback welcome!
- Add issue
- Install the dev environment (see below)
- Fork, add changes to master branch, ensure tests pass with full coverage and add a Pull Request
pip install pipenv
git clone https://github.com/davidhuser/dhis2.py
cd dhis2.py
pipenv install --dev
pipenv run tests
# install pre-commit hooks
pipenv run pre-commit install
# run type annotation check
pipenv run mypy dhis2
# run flake8 style guide enforcement
pipenv run flake8dhis2.py's source is provided under MIT license. See LICENCE for details.
- Copyright (c), 2020, David Huser