Genericised the Request class somewhat, added repr definitions
This commit is contained in:
Matthew Connelly
parent
d7e7528c10
commit
6f771fce0c
|
|
@ -1,9 +1,22 @@
|
|||
import typing
|
||||
# Environment var definitions
|
||||
ENV_URI='TCX_SITE_URI'
|
||||
ENV_AUTH_USER='TCX_API_AUTH_USERNAME'
|
||||
ENV_AUTH_PASS='TCX_API_AUTH_PASSWORD'
|
||||
|
||||
#This became a class so we can combine envvar initialisation with cred storage
|
||||
class Py3CXEnvironmentVars:
|
||||
Prefix='TCX'
|
||||
SiteComponent='SITE'
|
||||
AuthComponent='API_AUTH'
|
||||
EnvTemplate='{}_{}_{}'
|
||||
URI=EnvTemplate.format(Prefix, SiteComponent, 'URI')
|
||||
AUTH_USER=EnvTemplate.format(Prefix, AuthComponent, 'USERNAME')
|
||||
AUTH_PWD=EnvTemplate.format(Prefix, AuthComponent, 'PASSWORD')
|
||||
def __init__(self):
|
||||
from os import getenv
|
||||
self.uri=getenv(self.URI)
|
||||
self.auth_user=getenv(self.AUTH_USER)
|
||||
self.auth_pwd=getenv(self.AUTH_PWD)
|
||||
|
||||
#Base class for exceptions means one can 'except Py3CXException'
|
||||
#TODO: should this instead be a subclass of Py3CX? (except Py3CX.Exception?)
|
||||
class Py3CXException(Exception):
|
||||
pass
|
||||
class APIError(Py3CXException):
|
||||
|
|
@ -11,8 +24,9 @@ class APIError(Py3CXException):
|
|||
class ValidationError(Py3CXException):
|
||||
pass
|
||||
|
||||
#Request is a basic wrapper around python-requests to remove the need to always specify the API base uri, and to handle basic result validation on behalf of the caller
|
||||
class Request:
|
||||
from requests import ConnectionError, exceptions
|
||||
from requests import Response, ConnectionError, exceptions
|
||||
def __init__(self, uri, verify):
|
||||
from requests import Session
|
||||
self.uri=uri
|
||||
|
|
@ -20,10 +34,17 @@ class Request:
|
|||
self.last=None
|
||||
self.sess=Session()
|
||||
self.sess.verify=verify
|
||||
|
||||
def get(self, api, params=None, expect=200):
|
||||
def __repr__(self):
|
||||
return '<Request [%s,v%s]>' % (self.base,self.sess.verify)
|
||||
def _action(self, api, method, params, expect=200) -> 'Response':
|
||||
try:
|
||||
self.last=resp=self.sess.get(url=self.base.format(api), params=params)
|
||||
if isinstance(method, str):
|
||||
if method is 'GET':
|
||||
self.last=resp=self.sess.get(url=self.base.format(api), params=params, json=params)
|
||||
elif method is 'POST':
|
||||
self.last=resp=self.sess.post(url=self.base.format(api), json=params, params=params)
|
||||
else:
|
||||
self.last=resp=method(url=self.base.format(api), json=params, 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))
|
||||
|
|
@ -34,38 +55,36 @@ class Request:
|
|||
except ConnectionError as e:
|
||||
raise APIError("ConnectionError raised when communicating: %s" % str(e))
|
||||
return resp
|
||||
def get(self, api, params=None, expect=200) -> 'Response':
|
||||
return self._action(api, self.sess.get, params=params, expect=expect)
|
||||
def post(self, api, params, expect=200) -> 'Response':
|
||||
return self._action(api, self.sess.post, params=params, expect=expect)
|
||||
|
||||
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 self.exceptions.SSLError as e:
|
||||
raise APIError("TLS error returned when communicating; use tls_verify=False or check leaf certs: %s" % str(e))
|
||||
except self.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
|
||||
|
||||
#BasicObject is literally just an object - hacky way of making a variable addressable via var.prop rather than var['prop']
|
||||
class BasicObject(object):
|
||||
pass
|
||||
class APIObject(object):
|
||||
#APIObject is a basic object type that has some glue to reduce duplicate code
|
||||
class APIObject(BasicObject):
|
||||
def __init__(self, parent, api):
|
||||
self.tcx=parent
|
||||
self.api=api
|
||||
def __repr__(self):
|
||||
return '<APIObject [%s]>' % self.api
|
||||
#ReadOnlyObject is a basic object type that expects to be able to GET the given API endpoint and retrieve a JSON response
|
||||
class ReadOnlyObject(APIObject):
|
||||
def __repr__(self):
|
||||
return '<ReadOnlyObject [%s]>' % self.api
|
||||
def refresh(self, params={}):
|
||||
self._result=self.tcx.rq.get(self.api, params=params)
|
||||
self.active=self._result.json()
|
||||
#TransactionalObject is an object type that expects to retrieve details from an API via one endpoint, then use standard 3CX transactional endpoints for either cancelling the modify request, or submit modifications
|
||||
class TransactionalObject(APIObject):
|
||||
def __init__(self, parent, api, save='edit/save', discard='edit/cancel'):
|
||||
super().__init__(parent, api)
|
||||
self.save=save
|
||||
self.discard=discard
|
||||
def __repr__(self):
|
||||
return '<TransactionalObject [%s]>' % self.api
|
||||
def create(self, params):
|
||||
self._result=self.tcx.rq.post(self.api, params=params)
|
||||
self._session=self._result.json()['Id']
|
||||
|
|
@ -76,6 +95,7 @@ class TransactionalObject(APIObject):
|
|||
self.tcx.rq.post(self.discard, params={
|
||||
'Id':self._session})
|
||||
|
||||
#Main logic goes here
|
||||
class Py3CX:
|
||||
class _Call(ReadOnlyObject):
|
||||
def __init__(self, tcx, callid):
|
||||
|
|
@ -94,6 +114,8 @@ class Py3CX:
|
|||
self.state=res['Status']
|
||||
self.duration=res['Duration']
|
||||
self.since=res['LastChangeStatus']
|
||||
def __repr__(self):
|
||||
return '<Py3CX.Call [%s %s]>' % (self.id, self.state)
|
||||
def hangup(self):
|
||||
self.tcx.rq.post('activeCalls/drop', params={
|
||||
'Id': self.params})
|
||||
|
|
@ -103,6 +125,8 @@ class Py3CX:
|
|||
super().__init__(parent, 'ExtensionList/set')
|
||||
self.params=params
|
||||
self.load()
|
||||
def __repr__(self):
|
||||
return '<Py3CX.User [%s(%s)]>' % (self.params, self.sip_authid)
|
||||
def load(self):
|
||||
self.create(params={
|
||||
'Id':self.params})
|
||||
|
|
@ -126,6 +150,8 @@ class Py3CX:
|
|||
self.params=params
|
||||
if populate:
|
||||
self.load()
|
||||
def __repr__(self):
|
||||
return '<Py3CX.Extension [%s]>' % self.number
|
||||
def load(self):
|
||||
self.refresh(params=self.params)
|
||||
res=self._result
|
||||
|
|
@ -154,6 +180,8 @@ class Py3CX:
|
|||
self.tcx=tcx
|
||||
self.calllist=None
|
||||
self.extlist=None
|
||||
def __repr__(self):
|
||||
return '<Py3CX.System [%s]>' % self.tcx.cnf.uri
|
||||
def refresh_sysstat(self):
|
||||
sysstat=self.tcx.rq.get('SystemStatus').json()
|
||||
self.Status.exts_online=sysstat['ExtensionsRegistered']
|
||||
|
|
@ -239,29 +267,32 @@ class Py3CX:
|
|||
ret=[]
|
||||
for call in res:
|
||||
this=Py3CX._Call(self.tcx, call['Id'])
|
||||
this.id=call['Id']
|
||||
this.caller=call['Caller']
|
||||
this.callee=call['Callee']
|
||||
this.state=call['Status']
|
||||
this.duration=res['Duration']
|
||||
this.since=res['LastChangeStatus']
|
||||
this.duration=call['Duration']
|
||||
this.since=call['LastChangeStatus']
|
||||
ret.append(this)
|
||||
return ret
|
||||
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)
|
||||
self.__tcxsystem=None
|
||||
self.cnf=Py3CXEnvironmentVars()
|
||||
self._tcxsystem=None
|
||||
if uri is not None:
|
||||
self.cnf.uri=uri
|
||||
assert (self.cnf.uri is not None and self.cnf.uri.startswith('http')), "No or invalid URI specified, please provide via uri= or by setting environment variable %s" % self.cnf.URI
|
||||
self.rq=Request(uri=self.cnf.uri, verify=tls_verify)
|
||||
def __repr__(self):
|
||||
return '<Py3CX [%s@%s]>' % (self.cnf.auth_user, self.cnf.uri)
|
||||
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"
|
||||
if username is not None:
|
||||
self.cnf.auth_user=username
|
||||
if password is not None:
|
||||
self.cnf.auth_pwd=password
|
||||
assert (self.cnf.auth_user is not None and self.cnf.auth_pwd is not None), "Authentication information needed. Please pass username= and password= or define environment variables %s and %s" % (self.cnf.AUTH_USER, self.cnf.AUTH_PWD)
|
||||
rs=self.rq.post('login', params={
|
||||
'Username': self.uname,
|
||||
'Password': self.passw})
|
||||
'Username': self.cnf.auth_user,
|
||||
'Password': self.cnf.auth_pwd})
|
||||
self.rq.sess.headers.update({'x-xsrf-token':rs.cookies['XSRF-TOKEN']})
|
||||
@property
|
||||
def authenticated(self) -> bool:
|
||||
|
|
@ -273,7 +304,7 @@ class Py3CX:
|
|||
@property
|
||||
def system(self) -> 'Py3CX._PhoneSystem':
|
||||
assert self.authenticated, "Py3CX not authenticated yet!"
|
||||
if self.__tcxsystem is None:
|
||||
self.__tcxsystem=Py3CX._PhoneSystem(self)
|
||||
return self.__tcxsystem
|
||||
if self._tcxsystem is None:
|
||||
self._tcxsystem=Py3CX._PhoneSystem(self)
|
||||
return self._tcxsystem
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue