FAELIX
/
sipcentric
2
0
Fork 0
Code Issues Releases Wiki Activity

refactored completely

Browse Source
master
Marek Isalski 3 years ago
parent 016df8873b
commit 188c4d46c8
1 changed files with 195 additions and 112 deletions
Show Stats Download Patch File Download Diff File

307
src/sipcentric/__init__.py
Unescape Escape View File

@ -19,38 +19,32 @@ from .stream import Stream
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class Sipcentric(object): API_BASE = "https://pbx.sipcentric.com/api/v1"
def __init__(
self, ACCOUNTTYPE_BUSINESS = "BUSINESS"
username, ACCOUNTTYPE_RESIDENTIAL = "RESIDENTIAL"
password,
base="https://pbx.sipcentric.com/api/v1",
customer="me", class SipcentricException(Exception):
): pass
self.username = username # Account username
self.password = password # Account password class AuthenticationException(SipcentricException):
self.base = base # Base API URL (default: https://pbx.sipcentric.com/api/v1) pass
self.customer = customer # Customer (default: me)
# Resources class API(object):
self.account = Account(self) def __init__(self, username, password, base=None):
self.callBundles = CallBundles(self) self.username = username
self.recordings = Recordings(self) self.password = password
self.phoneBook = PhoneBook(self) self.base = API_BASE if base is None else base
self.timeIntervals = TimeIntervals(self)
self.endpoints = Endpoints(self)
self.phoneNumbers = PhoneNumbers(self)
self.sms = Sms(self)
self.creditStatus = CreditStatus(self)
self.calls = Calls(self)
self.sounds = Sounds(self)
self.outgoingCallerIds = OutgoingCallerIds(self)
self.stream = Stream(self)
def _request(self, uri, method="GET", data=None, params=None): def _request(self, uri, method="GET", data=None, params=None):
url = self.base + "/customers/" + self.customer + "/" + uri if uri.startswith("https://"):
auth = requests.auth.HTTPBasicAuth(self.username, self.password) # Basic auth url = uri
else:
url = self.base + uri
auth = requests.auth.HTTPBasicAuth(self.username, self.password)
print(method, url, params)
if method == "GET": if method == "GET":
if params: if params:
@ -91,7 +85,7 @@ class Sipcentric(object):
elif r.status_code == 401: elif r.status_code == 401:
raise AuthenticationException( raise AuthenticationException(
"We couldn't authenticate you with the API. Make sure you are using the correct credentials from the 'Web Users' section of the control panel. If you dont have an account, sign up for one at https://my.nimvelo.com/signup" "Could not authenticate you with the API. Make sure you are using the correct credentials from the 'Web Users' section of the control panel."
) )
return False return False
@ -101,125 +95,214 @@ class Sipcentric(object):
else: else:
raise Exception( raise Exception(
"HTTP Error " + str(r.status_code), "HTTP Error " + str(r.status_code),
"Something went wrong with the request", "Something went wrong with the request.",
) )
return False return False
def get(self, uri, params=None):
return self._request(uri, method="GET", params=params)
class Account(object): def getMany(self, uri, params=None, startPage=1, pageSize=20):
def __init__(self, parent): if params is None:
self.parent = parent params = {'pageSize': pageSize, 'page': startPage}
self.uri = "" # Not needed for the base of the customer else:
params['pageSize'] = pageSize
def get(self): params['page'] = startPage
return self.parent._request(self.uri)
resp = self._request(uri, "GET", params=params)
for item in resp.get('items', []):
class CallBundles(object): yield item
def __init__(self, parent): nextPage = resp.get('nextPage', None)
self.parent = parent while nextPage:
self.uri = "callbundles" resp = self._request(nextPage, "GET")
for item in resp.get('items', []):
yield item
nextPage = resp.get('nextPage', None)
def post(self, uri, data=None, params=None):
return self._request(uri, method="POST", data=data, params=params)
def patch(self, uri, data=None, params=None):
return self._request(uri, method="PATCH", data=data, params=params)
def put(self, uri, data=None, params=None):
return self._request(uri, method="PUT", data=data, params=params)
class APIObject(object):
def __init__(self, api, id=None, data=None):
self._api = api
self._id = id
if data:
self._id = data.get('id', None)
self._data = data
@classmethod
def makeUrl(cls, _id=None):
raise NotImplementedError()
@property
def data(self):
if self._data is None:
self._data = self._api.get(self.makeUrl())
return self._data
@data.setter
def set_data(self, data):
self._data = data
def create(self):
if self._id:
raise ValueError("%s ID %d already created" % (self.__class__.__name__, self._id))
if not self._data:
raise ValueError("No data associated with record.")
self._data = self._api.post(self.__class__.makeUrl(), data=self._data)
self._id = self._data.get('id')
return self._data
def update(self):
if not self._id:
raise ValueError("%s not yet created" % (self.__class__.__name__,))
if not self._data:
raise ValueError("No data associated with record.")
self._data = self._api.patch(self.makeUrl(), data=self._data)
self._id = self._data.get('id')
return self._data
@classmethod
def makeUrl(cls, id=None, parent=None):
if id:
path = "/%s/%d/" % (cls.URLPART, id)
else:
path = "/%s/" % (cls.URLPART)
if parent:
return (parent + path).replace("//", "/")
return path
def url(self, parent=None):
return self.__class__.makeUrl(self._id, parent)
class Account(APIObject):
@classmethod
def makeUrl(cls, id=None, parent=None):
if parent:
raise ValueError("Account should not have a parent")
if id:
raise NotImplementedError('There can be only one (Partner account)')
return "/"
def create(self):
raise NotImplementedError('You need to sign up as a new partner at https://www.simwood.com/partner/')
def customers(self):
for c in self._api.getMany(Customer.makeUrl()):
yield Customer(self._api, data=c)
class Customer(APIObject):
@classmethod
def makeUrl(cls, id=None, parent=None):
if parent:
raise ValueError("Customer should not have a parent")
if id == "me":
return "/customers/me/"
elif id:
return "/customers/%d/" % id
else:
return "/customers/"
def get(self): def endpoints(self):
return self.parent._request(self.uri) for c in self._api.getMany(Endpoint.makeUrl(parent=self.url())):
yield Endpoint(self._api, data=c)
def phonenumbers(self):
for c in self._api.getMany(PhoneNumber.makeUrl(parent=self.makurleUrl())):
yield PhoneNumber(self._api, data=c)
class Recordings(object): def calls(self, params=None):
def __init__(self, parent): for c in self._api.getMany(Call.makeUrl(parent=self.url()), params=params, pageSize=100):
self.parent = parent yield Call(self._api, data=c)
self.uri = "recordings"
def get(self): def callbundles(self):
return self.parent._request(self.uri) for c in self._api.getMany(CallBundle.makeUrl(parent=self.url())):
yield CallBundle(self._api, data=c)
def recordings(self):
for c in self._api.getMany(Recording.makeUrl(parent=self.url())):
yield Recording(self._api, data=c)
class PhoneBook(object): def phonebooks(self):
def __init__(self, parent): for c in self._api.getMany(PhoneBook.makeUrl(parent=self.url())):
self.parent = parent yield PhoneBook(self._api, data=c)
self.uri = "phonebook"
def get(self): def timeintervals(self):
return self.parent._request(self.uri) for c in self._api.getMany(TimeInterval.makeUrl(parent=self.url())):
yield TimeInterval(self._api, data=c)
def smss(self):
for c in self._api.getMany(Sms.makeUrl(parent=self.url())):
yield Sms(self._api, data=c)
class TimeIntervals(object): def sounds(self):
def __init__(self, parent): for c in self._api.getMany(Sound.makeUrl(parent=self.url())):
self.parent = parent yield Sound(self._api, data=c)
self.uri = "timeintervals"
def get(self): def outgoingcallerids(self):
return self.parent._request(self.uri) for c in self._api.getMany(OutgoingCallerId.makeUrl(parent=self.url())):
yield OutgoingCallerId(self._api, data=c)
def creditstatuses(self):
for c in self._api.getMany(CreditStatus.makeUrl(parent=self.url())):
yield CreditStatus(self._api, data=c)
class Endpoints(object):
def __init__(self, parent):
self.parent = parent
self.uri = "endpoints"
def get(self): class Endpoint(APIObject):
return self.parent._request(self.uri) URLPART = "endpoints"
class PhoneNumbers(object): class PhoneNumber(APIObject):
def __init__(self, parent): URLPART = "phonenumbers"
self.parent = parent
self.uri = "phonenumbers"
def get(self):
return self.parent._request(self.uri)
class Call(APIObject):
URLPART = "calls"
class Sms(object):
def __init__(self, parent):
self.parent = parent
self.uri = "sms"
def get(self): class CallBundle(APIObject):
return self.parent._request(self.uri) URLPART = "callbundles"
def post(self, to=None, _from=None, body=None):
data = {"type": "smsmessage", "to": to, "from": _from, "body": body}
return self.parent._request(self.uri, method="POST", data=data)
class Recording(APIObject):
URLPART = "recordings"
class CreditStatus(object):
def __init__(self, parent):
self.parent = parent
self.uri = "creditstatus"
def get(self): class PhoneBook(APIObject):
return self.parent._request(self.uri) URLPART = "phonebook"
class Calls(object): class TimeInterval(APIObject):
def __init__(self, parent): URLPART = "timeintervals"
self.parent = parent
self.uri = "calls"
def get(self):
return self.parent._request(self.uri)
class Sound(APIObject):
URLPART = "sounds"
class Sounds(object):
def __init__(self, parent):
self.parent = parent
self.uri = "sounds"
def get(self): class OutgoingCallerId(APIObject):
return self.parent._request(self.uri) URLPART = "outgoingcallerids"
class OutgoingCallerIds(object): class CreditStatus(APIObject):
def __init__(self, parent): URLPART = "creditstatus"
self.parent = parent
self.uri = "outgoingcallerids"
def get(self):
return self.parent._request(self.uri)
class Sms(APIObject):
URLPART = "sms"
class AuthenticationException(Exception): # def post(self, to=None, _from=None, body=None):
pass # data = {"type": "smsmessage", "to": to, "from": _from, "body": body}
# return self.parent._request(self.uri, method="POST", data=data)
if __name__ == "__main__": if __name__ == "__main__":

Write Preview
Loading…
Cancel
Save