py-3cx/Py3CX/__init__.py

# Environment var definitions
ENV_URI='TCX_SITE_URI'
ENV_AUTH_USER='TCX_API_AUTH_USERNAME'
ENV_AUTH_PASS='TCX_API_AUTH_PASSWORD'

class APIError(Exception):
	pass
class ValidationError(Exception):
	pass

class Request:
	from requests import ConnectionError, exceptions
	def __init__(self, uri, verify):
		from requests import Session
		self.uri=uri
		self.base="%s/api/{}" % uri
		self.last=None
		self.sess=Session()
		self.sess.verify=verify
	
	def get(self, api, params=None, expect=200):
		try:
			self.last=resp=self.sess.get(url=self.base.format(api), params=params)
			assert (resp.status_code==expect)
		except AssertionError:
			raise APIError("Assertion error when handling API response; response code was %s, but expected response was %s" % (resp.status_code, expect))
		except exceptions.SSLError as e:
			raise APIError("TLS error returned when communicating; use tls_verify=False or check leaf certs: %s" % str(e))
		except exceptions.BaseHTTPError as e:
			raise APIError("HTTP error raised when communicating: %s" % str(e))
		except ConnectionError as e:
			raise APIError("ConnectionError raised when communicating: %s" % str(e))
		return resp

	def post(self, api, params, expect=200):
		try:
			self.last=resp=self.sess.post(url=self.base.format(api), json=params)
			assert (resp.status_code==expect)
		except AssertionError:
			raise APIError("Assertion error when handling API response; response code was %s, but expected response was %s" % (resp.status_code, expect))
		except exceptions.SSLError as e:
			raise APIError("TLS error returned when communicating; use tls_verify=False or check leaf certs: %s" % str(e))
		except exceptions.BaseHTTPError as e:
			raise APIError("HTTP error raised when communicating: %s" % str(e))
		except ConnectionError as e:
			raise APIError("ConnectionError raised when communicating: %s" % str(e))
		except Exception as e:
			raise APIError("Other exception raised during API call: %s" % str(e))
		return resp

class ReadOnlyObject(object):
	def __init__(self, parent, api):
		self.tcx=parent
		self.api=api
	def refresh(self, params={}):
		self._result=self.tcx.rq.get(self.api, params=params)
		self.active=self._result.json()
class TransactionalObject(object):
	def __init__(self, parent, api, submit='edit/save', discard='edit/cancel'):
		self.tcx=parent
		self.api=api
		self.submit=submit
		self.discard=discard
	def create(self, params):
		self._result=self.tcx.rq.post(self.api, params=params)
		self._session=self._result.json()['Id']
		self.active=self._result.json()['ActiveObject']
	def submit(self, params):
		raise ValidationError("NotImplemented error")
	def cancel(self):
		self.tcx.rq.post(self.discard, params={
			'Id':self._session})

class User(TransactionalObject):
	def __init__(self, parent, params):
		super().__init__(parent, 'ExtensionList/set')
		self.params=params
		self.load()
	def load(self):
		self.create(params={
			'Id':self.params})
		parms=self.active
		self.id=parms['Id']
		self.enabled=not parms['Disabled']['_value']
		self.sip_id=parms['SIPId']['_value']
		#RecordCallsOption - enum
		self.callrecs=parms['RecordCalls']['selected']
		self.sip_host=parms['MyPhoneLocalInterface']['selected']
		self.sip_host_addrs=parms['MyPhoneLocalInterface']['possibleValues']
		self.sip_authid=parms['AuthId']['_value']
		self.sip_authpw=parms['AuthPassword']['_value']
		self.provision_uri=parms['MyPhoneProvLink']['_value']
		self.webpw=parms['AccessPassword']['_value']
		self.extension=Extension(self.tcx, self.params)
		self.cancel()
class Extension(ReadOnlyObject):
	def __init__(self, parent, params):
		super().__init__(parent, api='ExtensionList')
		self.params=params
		self.load()
	def load(self):
		self.refresh(params=self.params)
		res=self._result
		self.timestamp=res.headers.get('date')
		res=list(filter(lambda ext: ext['Number'] == self.params, res.json()['list']))
		assert (len(res)>0), "No extension found for: %s" % self.params
		assert (len(res)==1), "More than one extension found for %s" % self.params
		res=res[0]
		self.number=res['Number']
		self.firstname=res['FirstName']
		self.surname=res['LastName']
		self.name="%s %s" % (self.firstname, self.surname)
		self.dnd=res['DND']
		self.status=res['CurrentProfile']
		self.mail=res['Email']
		self.cli=res['OutboundCallerId']
		self.mobile=res['MobileNumber']
		self.online=res['IsRegistered']

class Py3CX:
	def __init__(self, uri=None, tls_verify=True):
		from os import getenv
		self.uri=uri if uri is not None and uri.startswith('http') else getenv(ENV_URI, None)
		assert (self.uri is not None and self.uri.startswith('http')), "Please provide URI"
		self.uname=getenv(ENV_AUTH_USER, None)
		self.passw=getenv(ENV_AUTH_PASS, None)
		self.rq=Request(uri=self.uri, verify=tls_verify)

	def authenticate(self, username=None, password=None):
		if username is not None and password is not None:
			self.uname=username
			self.passw=password
		assert (self.uname is not None and self.passw is not None), "Authentication information needed"
		rs=self.rq.post('login', params={
			'Username': self.uname,
			'Password': self.passw})
		self.rq.sess.headers.update({'x-xsrf-token':rs.cookies['XSRF-TOKEN']})

	@property
	def authenticated(self):
		try:
			self.rq.get('CurrentUser')
		except APIError:
			return False
		return True