refactor

Author: Reedyn

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "alma-python-packages"
-version = "1.4.13"
+version = "1.4.14"
 description = "Python module for making requests to the Alma API"
 readme = "README.md"
 requires-python = ">=3.14"
@@ -21,4 +21,4 @@ requires = ["hatchling"]
 build-backend = "hatchling.build"
 
 [tool.hatch.build.targets.wheel]
-packages = "alma_python_packages"
+package = "src/alma_python_packages"

src/alma_python_packages/__init__.py

@@ -0,0 +1,84 @@
+import xmljson
+from xml.etree.ElementTree import fromstring
+import re
+from collections import OrderedDict
+
+
+class AlmaAnalyticsException(Exception):
+    """Custom docstring"""
+
+
+class AlmaAnalyticsParser:
+
+    def __init__(self, i):
+        def __parse_analytics__(xml_string):
+            urn_schema = '{urn:schemas-microsoft-com:xml-analysis:rowset}'
+            w3_schema = '{http://www.w3.org/2001/XMLSchema}'
+
+            raw_data = xmljson.badgerfish.data(fromstring(xml_string))
+
+            def __get_column_names__():
+                snake_case_pattern = re.compile(r'(?<!^)(?=[A-Z])')
+
+                column_names = \
+                raw_data['report']['QueryResult']['ResultXml'][urn_schema + 'rowset'][w3_schema + 'schema'][
+                    w3_schema + 'complexType'][w3_schema + 'sequence'][w3_schema + 'element']
+
+                temp_column_names = []
+                for column in column_names[1:]:  # Remove first column since its just the integer
+                    for attribute, attribute_value in column.items():
+                        if attribute == '@{urn:saw-sql}columnHeading':
+                            temp_column_names.append(
+                                snake_case_pattern.sub('_', attribute_value).lower().replace(' ', ''))
+
+                return temp_column_names
+
+            def __get_rows__():
+                try:
+                    return raw_data['report']['QueryResult']['ResultXml'][urn_schema + 'rowset'][urn_schema + 'Row']
+                except KeyError:
+                    return []
+
+            column_names = __get_column_names__()
+            temp_table = []
+            rows = __get_rows__()
+            if type(rows) is list:  # More than one item in the list
+                for row in __get_rows__():
+                    row.popitem(last=False)  # Remove the integer column
+                    temp_row = OrderedDict()
+                    iter = 0
+                    for column, column_value in row.items():
+                        try:
+                            temp_row[column_names[int(column[-1:])-1]] = column_value['$']
+                        except ValueError:
+                            raise (AlmaAnalyticsException('Failed to load column number'))
+
+                    temp_table.append(temp_row)
+            elif type(rows) is OrderedDict:
+                rows.popitem(last=False)  # Remove the integer column
+                temp_row = OrderedDict()
+                iter = 0
+                for column, column_value in rows.items():
+                    try:
+                        temp_row[column_names[iter]] = column_value['$']
+                        iter = iter + 1
+                    except ValueError:
+                        raise (AlmaAnalyticsException('Failed to load column number'))
+
+                temp_table.append(temp_row)
+
+            return temp_table
+
+        self.list = __parse_analytics__(i)
+
+    def get_table(self):
+        return self.list
+
+    def get_column(self, column_name):
+        temp_list = []
+        try:
+            for row in self.list:
+                temp_list.append(row[column_name])
+            return temp_list
+        except IndexError:
+            return None

src/alma_python_packages/analytics_parser.py

@@ -0,0 +1,112 @@
+import httplib2 as http
+import xml.etree.ElementTree as ElementTree
+import json
+
+
+class AlmaAPIException(Exception):
+    """Custom docstring"""
+
+
+class AlmaAPI:
+
+    def __init__(self, api, api_key):
+        self.api_url = 'https://api-eu.hosted.exlibrisgroup.com/almaws/v1/' + api + '/'
+        self.api_key = '?apikey=' + api_key
+
+    def get(self, *, request=False, query_params=None):
+        if request is False:
+            return False
+
+        __query_params__ = ''
+        if query_params is not None:
+            for key, value in query_params.items():
+                __query_params__ += '&' + key + '=' + value
+
+        url = self.api_url + request + self.api_key + __query_params__
+
+        try:
+            (response, content) = http.Http().request(url)
+        except TimeoutError as e:
+            raise AlmaAPIException('GET - ' + str(e))
+
+        try:
+            if response.status != 200:
+                if query_params['format'] == 'json':
+                    error_data = json.loads(content)
+                    try:
+                        error_message = error_data['errorList']['error'][0]['errorMessage']
+                    except KeyError as e:
+                        error_message = 'Unknown Error'
+                    raise AlmaAPIException('GET - ' + str(response.status) + ' - ' + error_message)
+                else:
+                    root = ElementTree.ElementTree(ElementTree.fromstring(content)).getroot()
+                    error_message = root[1][0][1].text
+                    raise AlmaAPIException('GET - ' + str(response.status) + ' - ' + error_message)
+        except ElementTree.ParseError as e:
+            raise AlmaAPIException('GET - ' + str(e))
+
+        return content.decode('utf8')
+
+    def put(self, *, request=False, body=False, query_params=None, content_type='application/xml'):
+        if request is False or body is False:
+            return False
+
+        __query_params__ = ''
+        if query_params is not None:
+            for key, value in query_params.items():
+                __query_params__ += '&' + key + '=' + value
+
+        url = self.api_url + request + self.api_key + __query_params__
+        headers = {'Content-type': content_type}
+
+        try:
+            (response, content) = http.Http().request(url, 'PUT', headers=headers, body=body)
+        except TimeoutError as e:
+            raise AlmaAPIException('PUT - ' + str(e))
+
+        if response.status != 200:
+            if query_params['format'] == 'json':
+                error_data = json.loads(content)
+                try:
+                    error_message = error_data['errorList']['error'][0]['errorMessage']
+                except KeyError as e:
+                    error_message = 'Unknown Error'
+                raise AlmaAPIException('PUT - ' + str(response.status) + ' - ' + error_message)
+            else:
+                root = ElementTree.ElementTree(ElementTree.fromstring(content)).getroot()
+                error_message = root[1][0][1].text
+                raise AlmaAPIException('PUT - ' + str(response.status) + ' - ' + error_message)
+
+        return content.decode('utf8')
+
+    def post(self, *, request=False, body=False, query_params=None, content_type='application/xml'):
+        if request is False or body is False:
+            return False
+
+        __query_params__ = ''
+        if query_params is not None:
+            for key, value in query_params.items():
+                __query_params__ += '&' + key + '=' + value
+
+        url = self.api_url + request + self.api_key + __query_params__
+        headers = {'Content-type': content_type}
+
+        try:
+            (response, content) = http.Http().request(url, 'POST', headers=headers, body=body)
+        except TimeoutError as e:
+            raise AlmaAPIException('POST - ' + str(e))
+
+        if response.status != 200:
+            if query_params['format'] == 'json':
+                error_data = json.loads(content)
+                try:
+                    error_message = error_data['errorList']['error'][0]['errorMessage']
+                except KeyError as e:
+                    error_message = 'Unknown Error'
+                raise AlmaAPIException('POST - ' + str(response.status) + ' - ' + error_message)
+            else:
+                root = ElementTree.ElementTree(ElementTree.fromstring(content)).getroot()
+                error_message = root[1][0][1].text
+                raise AlmaAPIException('POST - ' + str(response.status) + ' - ' + error_message)
+
+        return content.decode('utf8')

src/alma_python_packages/api.py

@@ -0,0 +1,149 @@
+import xml.etree.ElementTree as ElementTree
+from io import BytesIO as bytes
+
+
+class AlmaMARCException(Exception):
+    pass
+
+
+class AlmaMARCRecord:
+
+    def __init__(self, input, *, file=False):
+
+        # Parse string or file into XML Object
+        def __parse__(i, f):
+            if f:
+                return ElementTree.parse(i)
+            else:
+                return ElementTree.ElementTree(ElementTree.fromstring(i))
+        self.xml = __parse__(input, file)
+
+        # Set root and type
+        root = self.xml.getroot()
+        self.type = root.tag
+
+        # Except if type is not Bib MARC or Holding Marc
+        if self.type != 'bib' and self.type != 'holding':
+            raise AlmaMARCException('Not a MARC XML Object')
+
+        # Map record fields to a list
+        self.fields = list(self.xml.findall('.//record/*'))
+
+        # Get the attributes of the XMl Object
+        def __get_attributes__(xml):
+            record = xml.find('.//record')
+            attributes = list(xml.findall('./'))
+            attributes.remove(record)
+
+            output = {}
+            for attribute in attributes:
+                output[attribute.tag] = attribute.text
+
+            return output
+
+        self.attributes = __get_attributes__(self.xml)
+
+    def get_fields(self, *, tag=None, field_type='datafield'):
+
+        array = []
+
+        for field in self.fields:
+
+            # If no tag is specified, grab all fields with the same type
+            if not tag:
+                if field.tag == field_type:
+                    marc_field = Field(field)
+                    array.append(marc_field)
+            else:
+                if field.attrib.get('tag') == tag and field.tag == field_type:
+                    marc_field = Field(field)
+                    array.append(marc_field)
+
+        return array
+
+    def add_field(self, *, tag, field_type='datafield', ind1='', ind2='', subfields=None):
+        record_root = self.xml.find('.//record')
+        new_field = ElementTree.Element(field_type, tag=tag, ind1=ind1, ind2=ind2)
+
+        if subfields is None:
+            subfields = {}
+        for code, value in subfields.items():
+            new_subfield = ElementTree.Element('subfield', code=code)
+            new_subfield.text = value
+            new_field.append(new_subfield)
+
+        self.fields.append(new_field)
+        record_root.append(new_field)
+        return self.get_fields(field_type=field_type, tag=tag)
+
+    def remove_fields(self, *, tag=False, field_type='datafield'):
+        record_root = self.xml.find('.//record')
+        removal_list = []
+        for field in self.fields:
+            if not tag:
+                if field.tag == field_type:
+                    removal_list.append(field)
+                    record_root.remove(field)
+            else:
+                if field.attrib.get('tag') == tag and field.tag == field_type:
+                    removal_list.append(field)
+                    record_root.remove(field)
+
+        self.fields = [e for e in self.fields if e not in removal_list]
+        return self.fields
+
+    def to_string(self):
+        # Write to fake file so that we can write our own XML declaration string.
+        fake_file = bytes()
+        self.xml.write(fake_file, encoding='utf-8', xml_declaration=False)
+        return '<?xml version="1.0" encoding="UTF-8" standalone="yes"?>' + fake_file.getvalue().decode('utf-8')
+
+
+class Field:
+
+    def __init__(self, reference):
+        self.__reference__ = reference
+        self.field_type = reference.tag
+        self.attributes = reference.attrib
+        self.value = reference.text if not list(reference) else None
+        self.subfields = []
+        for subfield in list(reference):
+            self.subfields.append(Subfield(subfield))
+
+    def get_subfield(self, code):
+        for subfield in self.subfields:
+            if subfield.code == code:
+                return subfield
+        return None
+
+    def add_subfield(self, *, code, value):
+        new_subfield = ElementTree.Element('subfield', code=code)
+        new_subfield.text = value
+        self.subfields.append(Subfield(new_subfield))
+        self.__reference__.append(new_subfield)
+        return self.get_subfield(code=code)
+
+    def get_subfields(self):
+        return self.subfields
+
+    def set_subfield_value(self, *, code, value):
+        for subfield in self.subfields:
+            if subfield.code == code:
+                subfield.set_value(value)
+                return subfield
+        return None
+
+
+class Subfield:
+    def __init__(self, reference):
+        self.__reference__ = reference
+        self.code = reference.attrib['code']
+        self.value = reference.text
+
+    def set_code(self, code):
+        self.code = code
+        self.__reference__.attrib['code'] = code
+
+    def set_value(self, value):
+        self.value = value
+        self.__reference__.text = value