Move last components of session layer to client and remove session completely

This commit is contained in:
clerie 2022-01-17 12:20:54 +01:00
parent 80fb8cd7cc
commit 8d2e94f5f3
3 changed files with 42 additions and 76 deletions

View File

@ -3,6 +3,5 @@
from . import client
from . import connection
from . import messages
from . import session
from .client import OMMClient2

View File

@ -1,6 +1,6 @@
#!/usr/bin/env python3
from .session import Session
from .connection import Connection
from . import messages
class OMMClient2:
@ -13,29 +13,53 @@ class OMMClient2:
:param host: Hostname or IP address of the OMM
:param username: Username
:param password: Password
:param session: A :class:`mitel_ommclient2.session.Session` object
:param port: Port where to access the API, if None, use default value
:param ommsync: If True login as OMM-Sync client. Some operations in OMM-Sync mode might lead to destroy DECT paring.
Usage::
>>> c = OMMClient2("omm.local", "admin", "admin")
>>> c.ping()
Use session for not implemented features::
Use request to send custom messages::
>>> r = s.session.request(mitel_ommclient2.messages.Ping())
To get more contol over the connection handling, initialize
:class:`mitel_ommclient2.session.Session` manually::
>>> s = mitel_ommclient2.session.Session("omm.local", "admin", "admin", port=12345)
>>> c = OMMClient2(session=s)
>>> r = s.request(mitel_ommclient2.messages.Ping())
"""
def __init__(self, host=None, username=None, password=None, session=None):
if session is None:
self.session = Session(host, username, password)
else:
self.session = session
def __init__(self, host, username, password, port=None, ommsync=False):
self._host = host
self._username = username
self._password = password
self._port = port
self._ommsync = ommsync
# prepare connect arguments
kwargs = {}
if self._port is not None:
kwargs["port"] = self._port
# Connect
self._connection = Connection(self._host, **kwargs)
self._connection.connect()
# Login
r = self.request(messages.Open(self._username, self._password, UserDeviceSyncClient=self._ommsync))
r.raise_on_error()
def request(self, request):
"""
Sends a request, waits for response and returns response
:param request: Request object
Usage::
>>> r = c.request(mitel_ommclient2.messages.Ping())
>>> r.name
'PingResp'
"""
return self._connection.request(request)
def get_account(self, id):
"""
@ -44,7 +68,7 @@ class OMMClient2:
:param id: User id
"""
r = self.session.request(messages.GetAccount(id))
r = self.request(messages.GetAccount(id))
r.raise_on_error()
if r.account is None:
return None
@ -56,7 +80,7 @@ class OMMClient2:
:param id: Device id
"""
r = self.session.request(messages.GetPPDev(ppn))
r = self.request(messages.GetPPDev(ppn))
r.raise_on_error()
if r.pp is None:
return None
@ -69,7 +93,7 @@ class OMMClient2:
Returns `True` when response is received.
"""
r = self.session.request(messages.Ping())
r = self.request(messages.Ping())
if r.errCode is None:
return True
return False

View File

@ -1,57 +0,0 @@
#!/usr/bin/env python3
from time import sleep
from . import connection
from . import messages
class Session:
"""
Synchronous API session handler
:param host: Hostname or IP address of OMM
:param username: Username
:param password: Password
:param port: Port
:param ommsync: If True login as OMM-Sync client. Some operations in OMM-Sync mode might lead to destroy DECT paring.
:param connection_class: One of :class:`mitel_ommclient2.connection.Connection` or :class:`mitel_ommclient2.connection.SSLConnection`
Usage::
>>> s = Session("omm.local", "admin", "admin")
>>> s.request(mitel_ommclient2.messages.Ping())
"""
def __init__(self, host, username, password, port=None, ommsync=False, connection_class=None):
self.host = host
self.username = username
self.password = password
self.port = port
self.ommsync = ommsync
self.connection_class = connection_class
if self.connection_class is None:
self.connection_class = connection.Connection
self._connection = None
self._ensure_connection()
def _ensure_connection(self):
"""
Make sure we are connected and logged in
"""
if self._connection is None:
kwargs = {}
if self.port is not None:
kwargs["port"] = self.port
self._connection = self.connection_class(self.host, **kwargs)
self._connection.connect()
# Login
r = self.request(messages.Open(self.username, self.password, UserDeviceSyncClient=self.ommsync))
r.raise_on_error()
def request(self, request):
return self._connection.request(request)