Refactor code structure for improved readability and maintainability
This commit is contained in:
@@ -1,7 +1,7 @@
|
||||
import re
|
||||
import xml.etree.ElementTree as ET
|
||||
from enum import Enum
|
||||
from typing import Dict, Iterable, List, Optional, Tuple, Union
|
||||
from typing import Any, Dict, Iterable, List, Optional, Tuple, Type, Union
|
||||
|
||||
import requests
|
||||
from requests.adapters import HTTPAdapter
|
||||
@@ -366,43 +366,74 @@ def book_from_marc(rec: MarcRecord) -> BookData:
|
||||
)
|
||||
|
||||
|
||||
class PicaSchema(Enum):
|
||||
TITLE = "pica.tit"
|
||||
CALLSIGN = "pica.abr"
|
||||
ALL = "pica.all"
|
||||
DATE_FIRST_CREATION = "pica.ser"
|
||||
DATE_LAST_MODIFIED = "pica.aed"
|
||||
ISBN = "pica.isb"
|
||||
ISSN = "pica.isn"
|
||||
ISMN = "pica.ism"
|
||||
PPN = "pica.ppn"
|
||||
AUTHOR = "pica.per"
|
||||
|
||||
|
||||
class ALMASchema(Enum):
|
||||
pass
|
||||
|
||||
|
||||
class DublinCoreSchema(Enum):
|
||||
pass
|
||||
|
||||
|
||||
class CQLSchema(Enum):
|
||||
pass
|
||||
|
||||
|
||||
class SWBSchema(Enum):
|
||||
URL = "https://sru.k10plus.de/opac-de-627!rec=1?version=1.1&operation=searchRetrieve&query={}&maximumRecords=100&recordSchema=marcxml"
|
||||
ARGSCHEMA = "pica."
|
||||
ARGSCHEMA = PicaSchema
|
||||
NAME = "SWB"
|
||||
|
||||
|
||||
class DNBSchema(Enum):
|
||||
URL = "https://services.dnb.de/sru/dnb?version=1.1&operation=searchRetrieve&query={}&maximumRecords=100&recordSchema=MARC21-xml"
|
||||
ARGSCHEMA = ""
|
||||
ARGSCHEMA = DublinCoreSchema
|
||||
NAME = "DNB"
|
||||
|
||||
|
||||
class KOBVSchema(Enum):
|
||||
URL = "https://sru.kobv.de/k2?version=1.1&operation=searchRetrieve&query={}&startRecord=1&maximumRecords=100&recordSchema=marcxml"
|
||||
ARGSCHEMA = "dc."
|
||||
ARGSCHEMA = DublinCoreSchema
|
||||
NAME = "KOBV"
|
||||
|
||||
|
||||
class HebisSchema(Enum):
|
||||
URL = "http://sru.hebis.de/sru/DB=2.1?query={}&version=1.1&operation=searchRetrieve&stylesheet=http%3A%2F%2Fsru.hebis.de%2Fsru%2F%3Fxsl%3DsearchRetrieveResponse&recordSchema=marc21&maximumRecords=100&startRecord=1&recordPacking=xml&sortKeys=LST_Y%2Cpica%2C0%2C%2C"
|
||||
ARGSCHEMA = "pica."
|
||||
ARGSCHEMA = PicaSchema
|
||||
NAME = "HEBIS"
|
||||
REPLACE = {" ": "+", "&": "%26", "=": "+%3D+"}
|
||||
|
||||
|
||||
class OEVKSchema(Enum):
|
||||
URL = "https://sru.k10plus.de/opac-de-627-2?version=1.1&operation=searchRetrieve&query={}&maximumRecords=100&recordSchema=marcxml"
|
||||
ARGSCHEMA = "pica."
|
||||
ARGSCHEMA = PicaSchema
|
||||
NAME = "OEVK"
|
||||
|
||||
|
||||
class HBZSchema(Enum):
|
||||
URL = "https://eu04.alma.exlibrisgroup.com/view/sru/49HBZ_NETWORK?version=1.2&operation=searchRetrieve&recordSchema=marcxml&query={}&maximumRecords=100&recordSchema=marcxml"
|
||||
ARGSCHEMA = "alma."
|
||||
ARGSCHEMA = ALMASchema
|
||||
NAME = "HBZ"
|
||||
|
||||
|
||||
class ArgumentSchema(Enum):
|
||||
TITLE = (
|
||||
"title",
|
||||
"tit",
|
||||
)
|
||||
|
||||
|
||||
RVK_ALLOWED = r"[A-Z0-9.\-\/]" # conservative char set typically seen in RVK notations
|
||||
|
||||
@@ -506,7 +537,32 @@ def find_newer_edition(
|
||||
return [best] if best else None
|
||||
|
||||
|
||||
class _Api:
|
||||
class QueryTransformer:
|
||||
def __init__(
|
||||
self, api_schema: Type[PicaSchema], arguments: Union[Iterable[str], str]
|
||||
):
|
||||
self.api_schema = api_schema
|
||||
if isinstance(arguments, str):
|
||||
self.arguments = [arguments]
|
||||
else:
|
||||
self.arguments = arguments
|
||||
|
||||
def transform(self) -> Dict[str, Any]:
|
||||
arguments: List[str] = []
|
||||
schema = self.api_schema
|
||||
print(schema.TITLE.name)
|
||||
for arg in self.arguments:
|
||||
if "=" not in arg:
|
||||
continue
|
||||
key, value = arg.split("=", 1)
|
||||
if hasattr(schema, key.upper()):
|
||||
api_key = getattr(schema, key.upper()).value
|
||||
arguments.append(f"{api_key}={value}")
|
||||
|
||||
return arguments
|
||||
|
||||
|
||||
class Api:
|
||||
def __init__(
|
||||
self, site: str, url: str, prefix: str, replace: Optional[Dict[str, str]] = None
|
||||
):
|
||||
@@ -585,99 +641,5 @@ class _Api:
|
||||
# Not implemented: depends on catalog front-end; return empty string for now
|
||||
return ""
|
||||
|
||||
|
||||
class SWB(_Api):
|
||||
def __init__(self):
|
||||
self.site = SWBSchema.NAME.value
|
||||
self.url = SWBSchema.URL.value
|
||||
self.prefix = SWBSchema.ARGSCHEMA.value
|
||||
super().__init__(self.site, self.url, self.prefix)
|
||||
|
||||
|
||||
class DNB(_Api):
|
||||
def __init__(self):
|
||||
self.site = DNBSchema.NAME.value
|
||||
self.url = DNBSchema.URL.value
|
||||
self.prefix = DNBSchema.ARGSCHEMA.value
|
||||
super().__init__(self.site, self.url, self.prefix)
|
||||
|
||||
|
||||
class KOBV(_Api):
|
||||
def __init__(self):
|
||||
self.site = KOBVSchema.NAME.value
|
||||
self.url = KOBVSchema.URL.value
|
||||
self.prefix = KOBVSchema.ARGSCHEMA.value
|
||||
super().__init__(self.site, self.url, self.prefix)
|
||||
|
||||
|
||||
class HEBIS(_Api):
|
||||
def __init__(self):
|
||||
self.site = HebisSchema.NAME.value
|
||||
self.url = HebisSchema.URL.value
|
||||
self.prefix = HebisSchema.ARGSCHEMA.value
|
||||
self.replace = HebisSchema.REPLACE.value
|
||||
super().__init__(self.site, self.url, self.prefix, self.replace)
|
||||
|
||||
|
||||
class OEVK(_Api):
|
||||
def __init__(self):
|
||||
self.site = OEVKSchema.NAME.value
|
||||
self.url = OEVKSchema.URL.value
|
||||
self.prefix = OEVKSchema.ARGSCHEMA.value
|
||||
super().__init__(self.site, self.url, self.prefix)
|
||||
|
||||
|
||||
class HBZ(_Api):
|
||||
"""
|
||||
Small wrapper of the SRU API used to retrieve data from the HBZ libraries
|
||||
|
||||
All fields are available [here](https://eu04.alma.exlibrisgroup.com/view/sru/49HBZ_NETWORK?version=1.2)
|
||||
|
||||
Schema
|
||||
------
|
||||
HBZSchema: <HBZSchema>
|
||||
query prefix: alma.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.site = HBZSchema.NAME.value
|
||||
self.url = HBZSchema.URL.value
|
||||
self.prefix = HBZSchema.ARGSCHEMA.value
|
||||
super().__init__(self.site, self.url, self.prefix)
|
||||
|
||||
def search(self, query_args: Union[Iterable[str], str]):
|
||||
arguments =
|
||||
|
||||
# async KVK class:
|
||||
class KVK:
|
||||
def __init__(self):
|
||||
self.k10plus = SWB()
|
||||
self.dnb = DNB()
|
||||
self.hebis = HEBIS()
|
||||
self.oevk = OEVK()
|
||||
self.hbz = HBZ()
|
||||
self.kobv = KOBV()
|
||||
|
||||
def close(self):
|
||||
self.k10plus.close()
|
||||
self.dnb.close()
|
||||
self.hebis.close()
|
||||
self.oevk.close()
|
||||
self.hbz.close()
|
||||
self.kobv.close()
|
||||
|
||||
def __del__(self):
|
||||
self.close()
|
||||
|
||||
# async def get_all(self, query_args: Union[Iterable[str], str]) -> Dict[str, List[BookData]]:
|
||||
async def get_all(
|
||||
self, query_args: Union[Iterable[str], str]
|
||||
) -> Dict[str, List[BookData]]:
|
||||
results = {}
|
||||
results["K10Plus"] = self.k10plus.getBooks(query_args)
|
||||
results["DNB"] = self.dnb.getBooks(query_args)
|
||||
results["HEBIS"] = self.hebis.getBooks(query_args)
|
||||
results["OEVK"] = self.oevk.getBooks(query_args)
|
||||
results["HBZ"] = self.hbz.getBooks(query_args)
|
||||
results["KOBV"] = self.kobv.getBooks(query_args)
|
||||
return results
|
||||
# def search(self, query_args: Union[Iterable[str], str]):
|
||||
# arguments =
|
||||
|
||||
Reference in New Issue
Block a user