Initial commit using urllib.

Author: amh-mw

.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

@@ -0,0 +1,130 @@
+import hashlib
+import json
+import logging
+import urllib.parse
+import urllib.request
+
+_logger = logging.getLogger(__name__)
+
+
+class Vtapi:
+    def __init__(self, url):
+        self.session_name = None
+        self.url = url + '/webservice.php'
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, type, value, traceback):
+        try:
+            self.logout()
+        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):
+        data = {
+            'operation': 'create',
+            'sessionName': self.session_name,
+            'elementType': module,
+            'element': json.dumps(values),
+        }
+        response = self._urlopen(self.url, data=data)
+        return self._result(response)
+
+    def download(self, id):
+        params = {
+            'operation': 'download',
+            'sessionName': self.session_name,
+            'id': id,
+        }
+        response = self._urlopen(self.url, params=params)
+        return self._result(response)
+
+    def listtypes(self):
+        params = {
+            'operation': 'listtypes',
+            'sessionName': self.session_name,
+        }
+        response = self._urlopen(self.url, params=params)
+        return self._result(response)
+
+    def login(self, username, accesskey):
+        token = self._getchallenge(username)
+        self.session_name = self._login(username, token, accesskey)
+
+    def logout(self):
+        if self.session_name:
+            data = {
+                'operation': 'logout',
+                'sessionName': self.session_name,
+            }
+            self._urlopen(self.url, data=data)
+            self.session_name = None
+
+    def query(self, query):
+        params = {
+            'operation': 'query',
+            'sessionName': self.session_name,
+            'query': query,
+        }
+        try:
+            response = self._urlopen(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):
+        params = {
+            'operation': 'getchallenge',
+            'username': username,
+        }
+        response = self._urlopen(self.url, params=params)
+        result = self._result(response)
+        token = result['token']
+        return token
+
+    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._urlopen(self.url, data=data)
+        body = self._result(response)
+        return body['sessionName']
+
+    def _result(self, response):
+        body = json.loads(response)
+        if not body['success']:
+            raise VtapiError(**body['error'], status=response.status_code)
+        return body['result']
+
+    def _urlopen(self, url, data=None, params=None):
+        if params:
+            url = url + '?' + urllib.parse.urlencode(params)
+        if data:
+            data = urllib.parse.urlencode(data).encode('utf-8')
+        with urllib.request.urlopen(url, data=data) as f:
+            return f.read().decode('utf-8')
+
+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}")