Initial commit.

.gitignore

+*.pyc
+*.pyo
+
+.idea
+bin

main.py

+from kivy.app import App
+from kivy.logger import Logger
+
+from openmrs import RESTConnection, RESTCommunication
+
+
+class RestApp(App):
+    def __init__(self, **kwargs):
+        super(RestApp, self).__init__(**kwargs)
+        self.openmrs_connection = RESTConnection('localhost', 8080, 'admin', 'Admin123')
+
+    def create_lincoln_hospital(self):
+        self.openmrs_connection.send_request('location', {
+            'q': 'Lincoln Hospital'
+        }, None, self.on_lincoln_hospital_queried, self.on_failure, self.on_failure)
+
+    def on_lincoln_hospital_queried(self, _, response):
+        matches = [result for result in response['results'] if result['display'] == 'Lincoln Hospital']
+        if len(matches) > 0:
+            print(f'Location "Lincoln Hospital" already exists: {matches[0]["uuid"]}')
+        else:
+            self.openmrs_connection.send_request('location', None, {
+                'name': 'Lincoln Hospital'
+            }, self.on_lincoln_hospital_created, self.on_failure, self.on_failure)
+
+    # noinspection PyMethodMayBeStatic
+    def on_lincoln_hospital_created(self, _, response):  # pylint: disable=no-self-use
+        print(f'Location "Lincoln Hospital" created: {response["uuid"]}')
+
+    def create_health_concept(self):
+        def coroutine(communication):
+            datatypes = yield from communication.get_uuids('conceptdatatype')
+            classes = yield from communication.get_uuids('conceptclass')
+            concepts = yield from communication.get_uuids('concept')
+            try:
+                numeric = datatypes['Numeric']
+                finding = classes['Finding']
+            except KeyError as exception:
+                communication.on_failure(None, exception)
+                return
+            health_concept_uuid = concepts.get('Health')
+            if health_concept_uuid is None:
+                health_concept = yield from communication.send_request('concept', post_parameters={
+                    'names': [{'name': 'Health', 'locale': 'en'}],
+                    'datatype': numeric,
+                    'conceptClass': finding,
+                })
+                health_concept_uuid = health_concept['uuid']
+                print(f'Concept "Health" created: {health_concept_uuid}')
+            else:
+                print(f'Concept "Health" already exists: {health_concept_uuid}')
+        RESTCommunication(coroutine, self.openmrs_connection, self.on_failure).run()
+
+    # noinspection PyMethodMayBeStatic
+    def on_failure(self, _, error):  # pylint: disable=no-self-use
+        Logger.error(f'RestApp: {error}')
+
+
+if __name__ == '__main__':
+    app = RestApp()
+    app.run()

openmrs.py

+import base64
+import json
+from urllib.parse import quote
+
+from kivy.clock import Clock
+from kivy.network.urlrequest import UrlRequest
+
+
+class RESTConnection(object):
+    def __init__(self, authority, port, username, password):
+        self.authority = authority
+        self.port = port
+        credentials = f'{username}:{password}'
+        credentials = base64.standard_b64encode(credentials.encode('UTF8')).decode('UTF8')
+        self.headers = {
+            'Authorization': f'Basic {credentials}',
+            'Content-type': 'application/json',
+        }
+
+    def construct_url(self, resource, get_parameters=None):
+        get_parameters = '&'.join(f'{quote(str(key))}={quote(str(value))}' for key, value in get_parameters.items()) \
+            if get_parameters is not None else ''
+        return f'http://{self.authority}:{self.port}/openmrs/ws/rest/v1/{resource}?{get_parameters}'
+
+    def send_request_by_url(self, url, post_parameters, on_success, on_failure, on_error):
+        UrlRequest(url, req_headers=self.headers,
+                   req_body=json.dumps(post_parameters) if post_parameters is not None else None,
+                   on_success=on_success, on_failure=on_failure, on_error=on_error)
+
+    def send_request(self, resource, get_parameters, post_parameters, on_success, on_failure, on_error):
+        url = self.construct_url(resource, get_parameters)
+        self.send_request_by_url(url, post_parameters, on_success, on_failure, on_error)
+
+
+class RESTCommunication(object):
+    def __init__(self, coroutine, rest_connection, on_failure):
+        self.generator = coroutine(self)
+        self.rest_connection = rest_connection
+
+        def fail(request, error):
+            self.generator = None
+            on_failure(request, error)
+        self.on_failure = fail
+
+    def send_request_by_url(self, url, post_parameters=None):
+        result = None
+
+        def on_success(_, response):
+            nonlocal result
+            result = response
+            Clock.schedule_once(self._resume)
+        self.rest_connection.send_request_by_url(url, post_parameters, on_success, self.on_failure, self.on_failure)
+        yield
+        return result
+
+    def send_request(self, resource, get_parameters=None, post_parameters=None):
+        url = self.rest_connection.construct_url(resource, get_parameters)
+        return (yield from self.send_request_by_url(url, post_parameters))
+
+    def get_uuids(self, resource):
+        result = {}
+        url = self.rest_connection.construct_url(resource)
+        while True:
+            response = yield from self.send_request_by_url(url)
+            result.update({record['display']: record['uuid'] for record in response['results']})
+            for link in response.get('links', []):
+                if link['rel'] == 'next':
+                    url = link['uri']
+                    break
+            else:
+                break
+        return result
+
+    def _resume(self, _=None):
+        try:
+            next(self.generator)
+        except StopIteration:
+            pass
+
+    def run(self):
+        self._resume()

rest.kv

+BoxLayout:
+    orientation: 'vertical'
+    Button:
+        text: 'Create Location "Lincoln Hospital"'
+        on_press: app.create_lincoln_hospital()
+    Button:
+        text: 'Create Concept "Health"'
+        on_press: app.create_health_concept()