Skip to content

Commit

Permalink
feat: convert all base classes from classforge to attrs
Browse files Browse the repository at this point in the history 
This is due to classforge, although an incredible class system, being a package
thats pretty much not used anywhere else, and not giving the impression of
being under active maintenance.
  • Loading branch information
@nigelm
nigelm committed Jan 31, 2022
1 parent 86d9930 commit 0132a21
Show file tree
Hide file tree
Showing 5 changed files with 131 additions and 69 deletions.
34 changes: 18 additions & 16 deletions broadworks_ocip/api.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,9 +10,10 @@
import socket
import sys
import uuid
from typing import Callable
from typing import Dict

from classforge import Class
from classforge import Field
import attr
from lxml import etree

import broadworks_ocip.base
Expand All @@ -25,7 +26,8 @@
VERBOSE_DEBUG = 9


class BroadworksAPI(Class):
@attr.s(slots=True, kw_only=True)
class BroadworksAPI:
"""
BroadworksAPI - A class encapsulating the Broadworks OCI-P API
Expand All @@ -47,19 +49,19 @@ class BroadworksAPI(Class):
"""

host: str = Field(type=str, required=True, mutable=False)
port: int = Field(type=int, default=2208, mutable=False)
username: str = Field(type=str, required=True, mutable=False)
password: str = Field(type=str, required=True, mutable=False)
logger = Field(type=logging.Logger)
authenticated: bool = Field(type=bool, default=False)
connect_timeout: int = Field(type=int, default=8)
command_timeout: int = Field(type=int, default=30)
socket = Field(type=socket.socket, default=None) # type: socket.socket
session_id: str = Field(type=str)
_despatch_table = Field(type=dict)

def on_init(self):
host: str = attr.ib()
port: int = attr.ib(default=2208)
username: str = attr.ib()
password: str = attr.ib()
logger: logging.Logger = attr.ib(default=None)
authenticated: bool = attr.ib(default=False)
connect_timeout: int = attr.ib(default=8)
command_timeout: int = attr.ib(default=30)
socket = attr.ib(default=None)
session_id: str = attr.ib(default=None)
_despatch_table: Dict[str, Callable] = attr.ib(default=None)

def __attrs_post_init__(self):
"""
Initialise the API object.
Expand Down
133 changes: 97 additions & 36 deletions broadworks_ocip/base.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,12 +6,15 @@
"""
import re
from collections import namedtuple
from typing import Any
from typing import Dict
from typing import Optional
from typing import Tuple

import attr
from classforge import Class
from classforge import Field
from lxml import etree

from broadworks_ocip.exceptions import OCIErrorAPISetup
from broadworks_ocip.exceptions import OCIErrorResponse


Expand Down Expand Up @@ -41,19 +44,32 @@ class ElementInfo:
is_table: bool = attr.ib(default=False)


class OCIType(Class):
@attr.s(slots=True, frozen=True, kw_only=True)
class OCIType:
"""
OCIType - Base type for all the OCI-P component classes
"""

# Namespace maps used for various XML build tasks
DEFAULT_NSMAP = {None: "", "xsi": "http://www.w3.org/2001/XMLSchema-instance"}
DOCUMENT_NSMAP = {None: "C", "xsi": "http://www.w3.org/2001/XMLSchema-instance"}
ERROR_NSMAP = {
"c": "C",
None: "",
"xsi": "http://www.w3.org/2001/XMLSchema-instance",
}
@classmethod
def _default_nsmap(cls):
return {None: "", "xsi": "http://www.w3.org/2001/XMLSchema-instance"}

@classmethod
def _document_nsmap(cls):
return {None: "C", "xsi": "http://www.w3.org/2001/XMLSchema-instance"}

@classmethod
def _error_nsmap(cls):
return {
"c": "C",
None: "",
"xsi": "http://www.w3.org/2001/XMLSchema-instance",
}

@classmethod
def _elements(cls) -> Tuple[ElementInfo, ...]:
raise OCIErrorAPISetup(message="_elements should be defined in the subclass.")

@property
def type_(self):
Expand Down Expand Up @@ -81,7 +97,7 @@ def etree_components_(self, name=None):
"""
if name is None:
name = self.type_
element = etree.Element(name, nsmap=self.DEFAULT_NSMAP)
element = etree.Element(name, nsmap=self._default_nsmap())
return self.etree_sub_components_(element)

def etree_sub_components_(self, element: "etree._Element"):
Expand All @@ -94,7 +110,7 @@ def etree_sub_components_(self, element: "etree._Element"):
Returns:
etree: etree.Element() for this class
"""
for sub_element in self._ELEMENTS:
for sub_element in self._elements():
value = getattr(self, sub_element.name)
if sub_element.is_array:
if value is not None:
Expand Down Expand Up @@ -127,15 +143,15 @@ def etree_sub_element_(
element,
sub_element.xmlname,
{"{http://www.w3.org/2001/XMLSchema-instance}nil": "true"},
nsmap=self.DEFAULT_NSMAP,
nsmap=self._default_nsmap(),
)
elif sub_element.is_table:
# any table should be a list of namedtuple elements
if type(value) is list and len(value) > 0:
elem = etree.SubElement(
element,
sub_element.xmlname,
nsmap=self.DEFAULT_NSMAP,
nsmap=self._default_nsmap(),
)
first = value[0]
for col in first._fields:
Expand All @@ -150,14 +166,14 @@ def etree_sub_element_(
elem = etree.SubElement(
element,
sub_element.xmlname,
nsmap=self.DEFAULT_NSMAP,
nsmap=self._default_nsmap(),
)
value.etree_sub_components_(elem)
else:
elem = etree.SubElement(
element,
sub_element.xmlname,
nsmap=self.DEFAULT_NSMAP,
nsmap=self._default_nsmap(),
)
if sub_element.type == bool:
elem.text = "true" if value else "false"
Expand Down Expand Up @@ -270,7 +286,7 @@ def build_from_etree_(cls, element: "etree._Element"):
results: Object instance for this class
"""
initialiser = {}
for elem in cls._ELEMENTS:
for elem in cls._elements():
if elem.is_array:
result = []
nodes = element.findall(elem.xmlname)
Expand All @@ -289,7 +305,31 @@ def build_from_etree_(cls, element: "etree._Element"):
# use that to build a new object
return cls(**initialiser)

def to_dict(self) -> Dict[str, Any]:
"""
Convert object to dict representation of itself
This was provided as part of the Classforge system, which we have moved away
from, so this is a local re-implementation. This is only used within the test
suite at present.
"""
elements = {}
for elem in self._elements():
value = getattr(self, elem.name)
if elem.is_table:
pass
elif elem.is_complex:
if elem.is_array:
value = [x.to_dict() for x in value]
else:
value = value.to_dict()
elif elem.is_array:
value = [x for x in value]
elements[elem.name] = value
return elements


@attr.s(slots=True, frozen=True, kw_only=True)
class OCICommand(OCIType):
"""
OCICommand - base class for all OCI Command (Request/Response) types
Expand All @@ -302,7 +342,7 @@ class OCICommand(OCIType):
there to give a known value for testing.
"""

session_id: str = Field(type=str, default="00000000-1111-2222-3333-444444444444")
session_id: str = attr.ib(default="00000000-1111-2222-3333-444444444444")

def build_xml_(self):
"""
Expand All @@ -317,11 +357,11 @@ def build_xml_(self):
root = etree.Element(
"{C}BroadsoftDocument",
{"protocol": "OCI"},
nsmap=self.DOCUMENT_NSMAP,
nsmap=self._document_nsmap(),
)
#
# add the session
session = etree.SubElement(root, "sessionId", nsmap=self.DEFAULT_NSMAP)
session = etree.SubElement(root, "sessionId", nsmap=self._default_nsmap())
session.text = self.session_id
#
# and the command
Expand Down Expand Up @@ -350,7 +390,7 @@ def build_xml_command_element_(self, root: "etree._Element"):
root,
"command",
{"{http://www.w3.org/2001/XMLSchema-instance}type": self.type_},
nsmap=self.DEFAULT_NSMAP,
nsmap=self._default_nsmap(),
)

@classmethod
Expand All @@ -368,7 +408,20 @@ def build_from_etree_non_parameters_(
if node is not None:
initialiser["session_id"] = node.text

def to_dict(self) -> Dict[str, Any]:
"""
Convert object to dict representation of itself
This was provided as part of the Classforge system, which we have moved away
from, so this is a local re-implementation. This is only used within the test
suite at present.
"""
elements = super().to_dict() # pick up the base object data
elements["session_id"] = self.session_id
return elements


@attr.s(slots=True, frozen=True, kw_only=True)
class OCIRequest(OCICommand):
"""
OCIRequest - base class for all OCI Command Request types
Expand All @@ -377,6 +430,7 @@ class OCIRequest(OCICommand):
pass


@attr.s(slots=True, frozen=True, kw_only=True)
class OCIResponse(OCICommand):
"""
OCIResponse - base class for all OCI Command Response types
Expand All @@ -393,19 +447,23 @@ def build_xml_command_element_(self, root: "etree._Element"):
root,
"command",
{"echo": "", "{http://www.w3.org/2001/XMLSchema-instance}type": self.type_},
nsmap=self.DEFAULT_NSMAP,
nsmap=self._default_nsmap(),
)


@attr.s(slots=True, frozen=True, kw_only=True)
class SuccessResponse(OCIResponse):
"""
The SuccessResponse is concrete response sent whenever a transaction is successful
and does not return any data.
"""

_ELEMENTS = () # type: ignore # type: Tuple[Tuple]
@classmethod
def _elements(cls) -> Tuple[ElementInfo, ...]:
return ()


@attr.s(slots=True, frozen=True, kw_only=True)
class ErrorResponse(OCIResponse):
"""
The ErrorResponse is concrete response sent whenever a transaction fails
Expand All @@ -415,18 +473,21 @@ class ErrorResponse(OCIResponse):
`OCIErrorResponse` exception is raised in `post_xml_decode_`.
"""

_ELEMENTS = (
ElementInfo("error_code", "errorCode", int),
ElementInfo("summary", "summary", str, is_required=True),
ElementInfo("summary_english", "summaryEnglish", str, is_required=True),
ElementInfo("detail", "detail", str),
ElementInfo("type", "type", str),
)
error_code = Field(type=int, required=False)
summary = Field(type=str, required=True)
summary_english = Field(type=str, required=True)
detail = Field(type=str, required=False)
type = Field(type=str, required=False)
error_code: Optional[int] = attr.ib(default=None)
summary: str = attr.ib()
summary_english: str = attr.ib()
detail: Optional[str] = attr.ib(default=None)
type: Optional[str] = attr.ib(default=None)

@classmethod
def _elements(cls) -> Tuple[ElementInfo, ...]:
return (
ElementInfo("error_code", "errorCode", int),
ElementInfo("summary", "summary", str, is_required=True),
ElementInfo("summary_english", "summaryEnglish", str, is_required=True),
ElementInfo("detail", "detail", str),
ElementInfo("type", "type", str),
)

def post_xml_decode_(self):
"""Raise an exception as this is an error"""
Expand All @@ -444,7 +505,7 @@ def build_xml_command_element_(self, root):
"echo": "",
"{http://www.w3.org/2001/XMLSchema-instance}type": "c:" + self.type_,
},
nsmap=self.ERROR_NSMAP,
nsmap=self._error_nsmap(),
)


Expand Down
14 changes: 14 additions & 0 deletions broadworks_ocip/exceptions.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ def __str__(self):
return f"{self.__class__.__name__}({self.message})"


@attr.s(slots=True, frozen=True)
class OCIErrorResponse(OCIError):
"""
Exception raised when an ErrorResponse is received and decoded.
Expand All @@ -32,6 +33,7 @@ class OCIErrorResponse(OCIError):
pass


@attr.s(slots=True, frozen=True)
class OCIErrorTimeOut(OCIError):
"""
Exception raised when nothing is head back from the server.
Expand All @@ -42,6 +44,7 @@ class OCIErrorTimeOut(OCIError):
pass


@attr.s(slots=True, frozen=True)
class OCIErrorUnknown(OCIError):
"""
Exception raised when life becomes too much for the software.
Expand All @@ -52,4 +55,15 @@ class OCIErrorUnknown(OCIError):
pass


@attr.s(slots=True, frozen=True)
class OCIErrorAPISetup(OCIError):
"""
Exception raised when life becomes too much for the software.
Subclass of OCIError()
"""

pass


# end

0 comments on commit 0132a21

Please sign in to comment.