dhis2.py 1.5.1

Python wrapper for DHIS2

A Python wrapper for DHIS2.

• Common HTTP operations (GET, POST, PUT, PATCH, DELETE)

• API paging for GETs

• SQL Views

• Server-side UID generation

• CSV/JSON file loading

• Defaults to JSON, supported GETs: XML, CSV, PDF, XLS

• requests as HTTP library

• logzero as drop-in logging library

• Supported and tested on Python 2.7, 3.4-3.7 and DHIS2 versions >= 2.25

Installation

Simply use pipenv (or pip):

pipenv install dhis2.py --user --upgrade

For instructions on installing Python / pip see The Hitchhiker’s Guide to Python.

Quickstart

Create an API object:

from dhis2 import Dhis

api = Dhis('play.dhis2.org/demo', 'admin', 'district', api_version=29, user_agent='myApp/0.1')

optional arguments:

• api_version: DHIS2 API version

• user_agent: submit your own User-Agent header

Then run requests on it:

r = api.get('organisationUnits/Rp268JB6Ne4', params={'fields': 'id,name'})

print(r.json())
# { "name": "Adonkia CHP", "id": "Rp268JB6Ne4" }

Get info about the DHIS2 instance

print(api.version)
# '2.29'

print(api.version_int)
# 29

print(api.revision)
# '17f7f0b'

print(api.api_url)
# 'https://play.dhis2.org/2.29/api/29'

Load authentication from file

Load from a auth JSON file in order to not store credentials in scripts. Must have the following structure:

{
  "dhis": {
    "baseurl": "https://play.dhis2.org/demo",
    "username": "admin",
    "password": "district"
  }
}

from dhis2 import Dhis

api = Dhis.from_auth_file('path/to/auth.json', api_version=29, user_agent='myApp/1.0')

If no argument is specified, it tries to find a file called dish.json in:

1. the DHIS_HOME environment variable

2. your Home folder

API paging

Paging for larger GET requests.

Two possible ways:

1. Process every page as they come in:

for page in api.get_paged('organisationUnits', page_size=100):
    print(page)
    # { "organisationUnits": [ {...}, {...} ] } (100 organisationUnits)

2. Load all pages before proceeding (this may take a long time) - to do this, do not use for and add merge=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.

Multiple params

You may submit params as a list of tuples instead when you need to pass multiple parameters with the same key, e.g.:

r = api.get('dataValueSets', params=[
        ('dataSet', 'pBOMPrpg1QX'), ('dataSet', 'BfMAe6Itzgt'),
        ('orgUnit', 'YuQRtpLP10I'), ('orgUnit', 'vWbkYPRmKyS'),
        ('startDate', '2013-01-01'), ('endDate', '2013-01-31')
    ]
)

or

r = api.get('dataValueSets', params={
    'dataSet': ['pBOMPrpg1QX', 'BfMAe6Itzgt'],
    'orgUnit': ['YuQRtpLP10I', 'vWbkYPRmKyS'],
    'startDate': '2013-01-01',
    'endDate': '2013-01-31'
})

SQL Views

Get SQL View data as if you’d open a CSV file, optimized for larger payloads:

# 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.

Generate UIDs

Get server-generated UIDs (not limited to 10000):

uids = api.generate_uids(20000)
print(uids)
# ['Rp268JB6Ne4', 'fa7uwpCKIwa', ... ]

GET other content types

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)

Load a JSON file

from dhis2 import load_json

json_data = load_json('/path/to/file.json')
print(json_data)
# { "id": ... }

Load a CSV file

from dhis2 import load_csv

for row in load_csv('/path/to/file.csv'):
    print(row)
    # { "id": ... }

# or for a normal list
data = list(load_csv('/path/to/file.csv'))

Logging

• 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.warn('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]

Exceptions

There are two exceptions:

• APIException: DHIS2 didn’t like what you requested. See the exception’s code, url and description.

• ClientException: Something didn’t work with the client not involving DHIS2.

They both inherit from Dhis2PyException.

Contribute

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