Skip to content

oarepo_oaipmh_harvester.transformers.oai_import

oarepo_oaipmh_harvester.transformers.oai_import

oai-import transformer - converts harvested XML with the target model's import.

The transformer takes the single child of the OAI <metadata> element and looks up an import of the target model whose oai_name equals (namespace, local name) of that element (imports are added to models with oarepo_model AddMetadataImport). The import's deserializer gets the element itself (deserialize_xml) if it supports it, otherwise its string serialization (deserialize). Deleted OAI records pass through with record = None.

Datastream configuration: oai-import{model:"<model name>"}.

If the deserializer raises (e.g. a marshmallow ValidationError), the record fails - the error is stored on its harvest record and the harvest goes on. See Readers, transformers and writers for an example.

OAIImportTransformer

Bases: BaseTransformer

Transformer for importing OAI-PMH records.

This transformers needs a model name to find the appropriate importer for the metadata format being imported.

Source code in oarepo_oaipmh_harvester/transformers/oai_import.py 
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
class OAIImportTransformer(BaseTransformer):
    """Transformer for importing OAI-PMH records.

    This transformers needs a model name to find the appropriate importer
    for the metadata format being imported.
    """

    def __init__(self, model: str, *_args: Any, **_kwargs: Any) -> None:
        """Initialize the OAI import transformer."""
        super().__init__()
        self.model = model

    def apply(self, stream_entry: StreamEntry, *_args: Any, **_kwargs: Any) -> StreamEntry:
        """Apply the transformation to the entry.

        Returns:
            StreamEntry: The transformed entry.

        Raises:
            TransformerError: In case of errors.

        """
        if not isinstance(stream_entry.entry, dict) or not isinstance(
            stream_entry.entry["record"],
            OAIRecord,
        ):
            raise TransformerError(
                "OAIImport transformer requires entries to be oaipmh_scythe Record instances, "
                f"it is {type(stream_entry)}."
            )
        if stream_entry.entry["record"].header.deleted:
            # deleted record, nothing to import
            return StreamEntry(
                entry={"record": None, "oai_record": stream_entry.entry["record"]},
                op_type=stream_entry.op_type,
            )
        # parse the metadata via the appropriate importer
        metadata_elem = self._get_metadata_element(stream_entry.entry["record"].xml)
        q = etree.QName(metadata_elem)
        namespace_uri = q.namespace
        local_name = q.localname
        imp = self._find_importer(namespace_uri, local_name)
        if xml_deserializer := getattr(imp.deserializer, "deserialize_xml", None):
            # if the deserializer can natively handle XML elements, use that
            data = xml_deserializer(metadata_elem)
        else:
            # otherwise convert to string and use the default deserialize method
            data = imp.deserializer.deserialize(etree.tostring(metadata_elem, encoding=str))
        return StreamEntry(
            entry={"record": data, "oai_record": stream_entry.entry["record"]},
            op_type=stream_entry.op_type,
        )

    def _find_importer(self, namespace_uri: str, local_name: str) -> Import:
        """Find the appropriate importer for the given OAI element."""
        for imp in current_runtime.models[self.model].imports:
            if not imp.oai_name:
                continue
            if imp.oai_name == (namespace_uri, local_name):
                return imp
        raise TransformerError(
            f"No import found for OAI element {{{namespace_uri}}}{local_name} in model {self.model}."
        )

    def _get_metadata_element(self, record_xml: etree._Element) -> etree._Element:
        """Extract the metadata element from an oaipmh_scythe Record XML.

        look for {http://www.openarchives.org/OAI/2.0/}record /
        {http://www.openarchives.org/OAI/2.0/}metadata and get the first child
        """
        oai_metadata_elem = record_xml.find("{http://www.openarchives.org/OAI/2.0/}metadata")
        if oai_metadata_elem is None:
            raise TransformerError("No metadata element found in OAI record.")
        children = [x for x in oai_metadata_elem if x.tag]
        if not children:
            raise TransformerError("No metadata child elements found in OAI record.")
        if len(children) > 1:
            raise TransformerError("Multiple metadata child elements found in OAI record.")
        return children[0]

__init__(model: str, *_args: Any, **_kwargs: Any) -> None

Initialize the OAI import transformer.

Source code in oarepo_oaipmh_harvester/transformers/oai_import.py 
42
43
44
45
def __init__(self, model: str, *_args: Any, **_kwargs: Any) -> None:
    """Initialize the OAI import transformer."""
    super().__init__()
    self.model = model

apply(stream_entry: StreamEntry, *_args: Any, **_kwargs: Any) -> StreamEntry

Apply the transformation to the entry.

Returns:

Name Type Description
StreamEntry StreamEntry

The transformed entry.

Raises:

Type Description
TransformerError

In case of errors.
Source code in oarepo_oaipmh_harvester/transformers/oai_import.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def apply(self, stream_entry: StreamEntry, *_args: Any, **_kwargs: Any) -> StreamEntry:
    """Apply the transformation to the entry.

    Returns:
        StreamEntry: The transformed entry.

    Raises:
        TransformerError: In case of errors.

    """
    if not isinstance(stream_entry.entry, dict) or not isinstance(
        stream_entry.entry["record"],
        OAIRecord,
    ):
        raise TransformerError(
            "OAIImport transformer requires entries to be oaipmh_scythe Record instances, "
            f"it is {type(stream_entry)}."
        )
    if stream_entry.entry["record"].header.deleted:
        # deleted record, nothing to import
        return StreamEntry(
            entry={"record": None, "oai_record": stream_entry.entry["record"]},
            op_type=stream_entry.op_type,
        )
    # parse the metadata via the appropriate importer
    metadata_elem = self._get_metadata_element(stream_entry.entry["record"].xml)
    q = etree.QName(metadata_elem)
    namespace_uri = q.namespace
    local_name = q.localname
    imp = self._find_importer(namespace_uri, local_name)
    if xml_deserializer := getattr(imp.deserializer, "deserialize_xml", None):
        # if the deserializer can natively handle XML elements, use that
        data = xml_deserializer(metadata_elem)
    else:
        # otherwise convert to string and use the default deserialize method
        data = imp.deserializer.deserialize(etree.tostring(metadata_elem, encoding=str))
    return StreamEntry(
        entry={"record": data, "oai_record": stream_entry.entry["record"]},
        op_type=stream_entry.op_type,
    )