Added PAT support

dhis2/api.py

@@ -35,16 +35,20 @@ class Api(object):
     def __init__(
         self,
         server: str,
-        username: str,
-        password: str,
+        username: str = None,
+        password: str = None,
+        api_access_token: str = None,
         api_version: Union[int, str] = None,
         user_agent: str = None,
+        validate: bool = False,
+        deny_self_signed_certificate: bool = True
     ) -> None:
         """
 
         :param server: baseurl, e.g. 'play.dhis2.org/demo'
         :param username: DHIS2 username
         :param password: DHIS2 password
+        :param api_access_token: DHIS2 private api token
         :param api_version: optional, creates a url like /api/29/schemas
         :param user_agent: optional, add user-agent to header. otherwise it uses requests' user-agent.
         """
@@ -61,10 +65,21 @@ def __init__(
         self.api_version = api_version
 
         self.session = requests.Session()
-        self.username = username
-        self.session.auth = (self.username, password)
+        self.session.verify = deny_self_signed_certificate
+        if password is not None:
+            self.username = username
+            self.session.auth = (self.username, password)
+        elif api_access_token is not None:
+            self.session.headers["Authorization"] = f"ApiToken {api_access_token}"
+            self.username = "<You are using PAT>"
+        else:
+            raise ClientException("You need to define password or api_access_token in API construction.")
         if user_agent:
             self.session.headers["user-agent"] = user_agent
+        if validate:
+            self.username = self.get("me", params=[("fields","username")]).json()['username']
+            print(self.username)
+
 
     def get_base_url(self) -> Optional[str]:
         return self._base_url
@@ -236,9 +251,9 @@ def _validate_request(
                 [isinstance(elem, tuple) for elem in params]
             ):
                 raise ClientException("`params` list must all be tuples")
-        if data and not isinstance(data, dict):
+        if data and not isinstance(data, (dict, list)):
             raise ClientException(
-                "`data` must be a dict, not {}".format(data.__class__.__name__)  # type: ignore
+                "`data` must be a dict or list, not {}".format(data.__class__.__name__)  # type: ignore
             )
 
     def _make_request(
@@ -275,7 +290,11 @@ def _make_request(
             r = self.session.put(url=url, json=data, params=params, timeout=timeout)
 
         elif method == "patch":
-            r = self.session.patch(url=url, json=data, params=params, timeout=timeout)
+            self.session.headers["Content-Type"] = "application/json-patch+json"
+            try:
+                r = self.session.patch(url=url, json=data, params=params, timeout=timeout)
+            finally:
+                del self.session.headers["Content-Type"]
 
         elif method == "delete":
             r = self.session.delete(url=url, params=params, timeout=timeout)

examples/1_update_dataelements.py

@@ -7,7 +7,8 @@
 """
 
 # Create a Api object
-api = Api("play.dhis2.org/dev", "admin", "district")
+api = Api("play.im.dhis2.org/dev", "admin", "district")
+# api = Api("play.im.dhis2.org/dev", api_access_token="d2p_Y0WIFJZx9AFISuUS1EC8tCOlqUZy5S6QRwEiZZeFh3P33rl1wa", validate=True)
 
 # setup the logger
 setup_logger()

setup.py

@@ -71,7 +71,7 @@ def run(self):
     install_requires=requirements,
     license=about["__license__"],
     zip_safe=False,
-    classifiers=(  # https://pypi.org/pypi?%3Aaction=list_classifiers
+    classifiers=[  # https://pypi.org/pypi?%3Aaction=list_classifiers
         "License :: OSI Approved :: MIT License",
         "Intended Audience :: Developers",
         "Development Status :: 5 - Production/Stable",
@@ -80,6 +80,6 @@ def run(self):
         "Programming Language :: Python :: 3.6",
         "Programming Language :: Python :: 3.7",
         "Programming Language :: Python :: 3.8",
-    ),
+    ],
     cmdclass={"publish": PublishCommand},
 )