diff --git a/Py3CX/__init__.py b/Py3CX/__init__.py index d51d88f..eb3c52d 100644 --- a/Py3CX/__init__.py +++ b/Py3CX/__init__.py @@ -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 '' % (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 '' % 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 '' % 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 '' % 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 '' % (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 '' % (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 '' % 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 '' % 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 '' % (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