From b0f059bc3c7696d1c5d1083f77a43586ab879c3e Mon Sep 17 00:00:00 2001 From: Adam Heinz Date: Tue, 19 Nov 2024 09:14:49 -0500 Subject: [PATCH] Initial commit using urllib. --- .gitignore | 1 + test.sh | 17 ++++++ tests/__init__.py | 1 + tests/test_vtiger.py | 25 +++++++++ vtiger.py | 131 +++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 175 insertions(+) create mode 100644 .gitignore create mode 100755 test.sh create mode 100644 tests/__init__.py create mode 100644 tests/test_vtiger.py create mode 100644 vtiger.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..bee8a64 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +__pycache__ diff --git a/test.sh b/test.sh new file mode 100755 index 0000000..628fdf6 --- /dev/null +++ b/test.sh @@ -0,0 +1,17 @@ +#!/bin/sh -e +usage() { + echo "Usage: $0 [-h]" 1>&2; + echo " -h Display this help message." 1>&2; + exit 1; +} + +while getopts "hi" opt; do + case $opt in + *) + usage + ;; + esac +done + +PYTHON=python3.9 +$PYTHON -m unittest diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..b14bd9b --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +from . import test_vtiger diff --git a/tests/test_vtiger.py b/tests/test_vtiger.py new file mode 100644 index 0000000..720e2cd --- /dev/null +++ b/tests/test_vtiger.py @@ -0,0 +1,25 @@ +import os +import unittest + +from vtiger 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]) diff --git a/vtiger.py b/vtiger.py new file mode 100644 index 0000000..0241e06 --- /dev/null +++ b/vtiger.py @@ -0,0 +1,131 @@ +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}")