Skip to content

Commit

Permalink
feat: change base classes to be a custom object
Browse files Browse the repository at this point in the history
This is to overcome the startup delay when using attrs
  • Loading branch information
nigelm committed Feb 3, 2022
1 parent 7360a7f commit 6d89b0d
Show file tree
Hide file tree
Showing 3 changed files with 88 additions and 34 deletions.
68 changes: 55 additions & 13 deletions broadworks_ocip/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,14 +8,15 @@
from collections import namedtuple
from typing import Any
from typing import Dict
from typing import Optional
from typing import Tuple

import attr
from lxml import etree

from broadworks_ocip.exceptions import OCIErrorAPISetup
from broadworks_ocip.exceptions import OCIErrorAttributeMissing
from broadworks_ocip.exceptions import OCIErrorResponse
from broadworks_ocip.exceptions import OCIErrorUnexpectedAttribute


@attr.s(slots=True, frozen=True)
Expand Down Expand Up @@ -44,12 +45,44 @@ class ElementInfo:
is_table: bool = attr.ib(default=False)


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

__slots__ = ["_frozen"]

def __init__(self, **kwargs):
for elem in self._elements():
if elem.name in kwargs:
setattr(self, elem.name, kwargs[elem.name])
if elem.is_required and kwargs[elem.name] is None:
raise OCIErrorAttributeMissing(
message=f"Required attribute {elem.name} is missing",
)
del kwargs[elem.name]
elif elem.is_required:
raise OCIErrorAttributeMissing(
message=f"Required attribute {elem.name} is missing",
)
else:
setattr(self, elem.name, None)
if kwargs:
raise OCIErrorUnexpectedAttribute(
message=f"Unexpected attribute(s) {kwargs.keys()}",
)
self._frozen = True

def __delattr__(self, *args, **kwargs):
if hasattr(self, "_frozen"):
raise AttributeError("This object is frozen!")
object.__delattr__(self, *args, **kwargs)

def __setattr__(self, *args, **kwargs):
if hasattr(self, "_frozen"):
raise AttributeError("This object is frozen!")
object.__setattr__(self, *args, **kwargs)

# Namespace maps used for various XML build tasks
@classmethod
def _default_nsmap(cls):
Expand Down Expand Up @@ -329,7 +362,6 @@ def to_dict(self) -> Dict[str, Any]:
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 @@ -342,7 +374,15 @@ class OCICommand(OCIType):
there to give a known value for testing.
"""

session_id: str = attr.ib(default="00000000-1111-2222-3333-444444444444")
__slots__ = ["_frozen", "session_id"]

def __init__(
self,
session_id: str = "00000000-1111-2222-3333-444444444444",
**kwargs,
):
self.session_id = session_id
super().__init__(**kwargs)

def build_xml_(self):
"""
Expand Down Expand Up @@ -421,7 +461,6 @@ def to_dict(self) -> Dict[str, Any]:
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 @@ -430,7 +469,6 @@ 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 @@ -451,19 +489,19 @@ def build_xml_command_element_(self, root: "etree._Element"):
)


@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.
"""

__slots__ = ["_frozen", "session_id"]

@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 @@ -473,11 +511,15 @@ class ErrorResponse(OCIResponse):
`OCIErrorResponse` exception is raised in `post_xml_decode_`.
"""

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)
__slots__ = [
"error_code",
"summary",
"summary_english",
"detail",
"type",
"_frozen",
"session_id",
]

@classmethod
def _elements(cls) -> Tuple[ElementInfo, ...]:
Expand Down
22 changes: 22 additions & 0 deletions broadworks_ocip/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -66,4 +66,26 @@ class OCIErrorAPISetup(OCIError):
pass


@attr.s(slots=True, frozen=True)
class OCIErrorAttributeMissing(OCIError):
"""
Exception raised when a required attribute is missing.
Subclass of OCIError()
"""

pass


@attr.s(slots=True, frozen=True)
class OCIErrorUnexpectedAttribute(OCIError):
"""
Exception raised when additional elements passed to __init__
Subclass of OCIError()
"""

pass


# end
32 changes: 11 additions & 21 deletions utilities/process_schema.py
Original file line number Diff line number Diff line change
Expand Up @@ -103,25 +103,18 @@ def write_elements(file, elements):
file.write(" return ()\n")


def write_attribute(file, element):
param_str = "" if element["is_required"] else "default=None"
if element["is_array"]:
mytype = f'List[{ element["type"] }]'
elif element["type"] in ("str", "bool", "int"):
mytype = element["type"]
else:
mytype = '"' + element["type"] + '"'
outstr = " " * 4 + element["name"] + ": " + mytype + " = attr.ib(" + param_str + ")"
file.write(outstr + "\n")
def write_slots(file, elements, thing):
file.write(" __slots__ = [\n")
for element in elements.values():
file.write(f' "{element["name"]}",\n')
if thing != "OCIType":
file.write(' "session_id",\n')
file.write(' "_frozen",\n')
file.write(" ]\n\n")


def process_class_elements(file, elements):
count = 0
for element in elements.values():
write_attribute(file, element)
count += 1
if count > 0:
file.write("\n")
def process_class_elements(file, elements, thing):
write_slots(file, elements, thing)
write_elements(file, elements)


Expand Down Expand Up @@ -197,11 +190,10 @@ def process_documentation(file, xsd_component, elements):


def process_thing(file, xsd_component, thing, prefix=""):
file.write("@attr.s(slots=True, frozen=True, kw_only=True)\n")
file.write(f"class {xsd_component.name}({thing}):\n")
elements = build_element_hash(xsd_component, prefix)
process_documentation(file, xsd_component, elements)
process_class_elements(file, elements)
process_class_elements(file, elements, thing)
file.write("\n\n")


Expand Down Expand Up @@ -259,9 +251,7 @@ def open_output_files():
out.write("# Do not edit as changes will be overwritten.\n")
out.write(f"# Generated on {generation_time.isoformat()}\n")
out.write("# fmt: off\n")
out.write("from typing import List\n")
out.write("from typing import Tuple\n\n")
out.write("import attr\n\n")
if thing in ("request", "response"):
out.write("import broadworks_ocip.types as OCI\n")
out.write("from .base import ElementInfo as E\n")
Expand Down

0 comments on commit 6d89b0d

Please sign in to comment.