Switch from urllib3 to requests.

.gitignore

@@ -0,0 +1,2 @@
+__pycache__
+.venv

requirements.txt

@@ -0,0 +1 @@
+requests

test.sh

@@ -0,0 +1,29 @@
+#!/bin/sh -e
+usage() {
+    echo "Usage: $0 [-h]" 1>&2;
+    echo "  -h Display this help message." 1>&2;
+    echo "  -i Initialize virtual environment." 1>&2;
+    exit 1;
+}
+
+while getopts "hi" opt; do
+  case $opt in
+    i)
+      i=1
+      ;;
+    *)
+      usage
+      ;;
+  esac
+done
+
+PYTHON=python3.9
+
+if [ -n "$i" ]; then
+    $PYTHON -m venv .venv
+fi
+source .venv/bin/activate
+if [ -n "$i" ]; then
+    $PYTHON -m pip install -r requirements.txt
+fi
+$PYTHON -m unittest

tests/__init__.py

@@ -0,0 +1 @@
+from . import test_vtwsclib

tests/test_vtwsclib.py

@@ -0,0 +1,25 @@
+import os
+import unittest
+
+from vtwsclib import Vtapi
+
+
+class TestVtapi(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        if not all(key in os.environ for key in ['VTIGER_HOST', 'VTIGER_USER', 'VTIGER_PASS']):
+            cls.skipTest("environment variables not configured")
+
+    def test_login(self):
+        with Vtapi(os.environ['VTIGER_HOST']) as api:
+            api.login(os.environ['VTIGER_USER'], os.environ['VTIGER_PASS'])
+
+    def test_count(self):
+        with Vtapi(os.environ['VTIGER_HOST']) as api:
+            api.login(os.environ['VTIGER_USER'], os.environ['VTIGER_PASS'])
+            self.assertGreater(api.count('Quotes'), 0)
+
+    def test_retrieve(self):
+        with Vtapi(os.environ['VTIGER_HOST']) as api:
+            api.login(os.environ['VTIGER_USER'], os.environ['VTIGER_PASS'])
+            self.assertIn('website', api.retrieve('CompanyDetails')[0])

vtwsclib.py

@@ -2,16 +2,15 @@
 import json
 import logging
 
-import urllib3
+import requests
 
-_http = urllib3.PoolManager()
 _logger = logging.getLogger(__name__)
 
 
 class Vtapi:
     def __init__(self, url):
-        self.session = None
-        self.user_id = None
+        self.session = requests.Session()
+        self.session_name = None
         self.url = url + '/webservice.php'
 
     def __enter__(self):
@@ -23,82 +22,103 @@ def __exit__(self, type, value, traceback):
         finally:
             pass
 
+    def count(self, module):
+        query = f"select count(*) from {module};"
+        result = self.query(query)
+        return int(result[0]['count'])
+
     def create(self, module, values):
-        fields = {
+        data = {
             'operation': 'create',
-            'sessionName': self.session,
+            'sessionName': self.session_name,
             'elementType': module,
             'element': json.dumps(values),
         }
-        return self._request('POST', fields)
+        response = self.session.post(self.url, data=data)
+        return self._result(response)
 
     def download(self, id):
-        fields = {
+        params = {
             'operation': 'download',
-            'sessionName': self.session,
+            'sessionName': self.session_name,
             'id': id,
         }
-        return self._request('GET', fields)
+        response = self.session.get(self.url, params=params)
+        return self._result(response)
 
     def listtypes(self):
-        fields = {
+        params = {
             'operation': 'listtypes',
-            'sessionName': self.session,
+            'sessionName': self.session_name,
         }
-        return self._request('GET', fields)
+        response = self.session.get(self.url, params=params)
+        return self._result(response)
 
     def login(self, username, accesskey):
         token = self._getchallenge(username)
-        hasher = hashlib.md5()
-        hasher.update(token.encode('utf-8'))
-        hasher.update(accesskey.encode('utf-8'))
-        fields = {
-            'operation': 'login',
-            'username': username,
-            'accessKey': hasher.hexdigest(),
-        }
-        result = self._request('POST', fields=fields)
-        self.session, self.user_id = result['sessionName'], result['userId']
+        self.session_name = self._login(username, token, accesskey)
 
     def logout(self):
-        if self.session:
-            fields = {
+        if self.session_name:
+            data = {
                 'operation': 'logout',
-                'sessionName': self.session,
+                'sessionName': self.session_name,
             }
-            self._request('POST', fields)
-            self.session = None
-            self.user_id = None
+            self.session.post(self.url, data=data)
+            self.session_name = None
 
     def query(self, query):
-        fields = {
+        params = {
             'operation': 'query',
-            'sessionName': self.session,
+            'sessionName': self.session_name,
             'query': query,
         }
-        return self._request('GET', fields)
-
-    def retrieve(self, type, limit, offset):
-        query = 'select * from {} limit {}, {};'.format(type, offset, limit),
+        try:
+            response = self.session.get(self.url, params=params)
+            return self._result(response)
+        except:
+            _logger.error("failed to query '%s'", query)
+            raise
+
+    def retrieve(self, module, limit=0, offset=0):
+        query = f"select * from {module};"
+        if limit or offset:
+            query = query[:-1] + f" limit {offset}, {limit};"
         return self.query(query)
 
     def _getchallenge(self, username):
-        fields = {
+        params = {
             'operation': 'getchallenge',
             'username': username,
         }
-        result = self._request('GET', fields)
-        return result['token']
+        response = self.session.get(self.url, params=params)
+        result = self._result(response)
+        token = result['token']
+        return token
 
-    def _request(self, method, fields):
-        _logger.debug('request %s', fields)
-        response = _http.request(method, self.url, fields=fields, retries=False)
-        return self._result(response)
+    def _login(self, username, token, accesskey):
+        hasher = hashlib.md5()
+        hasher.update(token.encode('utf-8'))
+        hasher.update(accesskey.encode('utf-8'))
+        data = {
+            'operation': 'login',
+            'username': username,
+            'accessKey': hasher.hexdigest(),
+        }
+        response = self.session.post(self.url, data=data)
+        body = self._result(response)
+        return body['sessionName']
 
     def _result(self, response):
-        data = response.data.decode('utf-8')
-        _logger.debug('response %s', data)
-        body = json.loads(data)
+        body = response.json()
         if not body['success']:
-            raise Exception(body['error']['message'])
+            raise VtapiError(**body['error'], status=response.status_code)
         return body['result']
+
+
+class VtapiError(Exception):
+    def __init__(self, code, message, status=None):
+        self.code = code
+        self.message = message
+        self.status = status
+        super().__init__(f"{self.code}: {self.message}")