Skip to content

oarepo_oaipmh_harvester.datastream

oarepo_oaipmh_harvester.datastream

Harvest datastream that never stops because of a single record.

The invenio-vocabularies DataStream catches only TransformerError / WriterError - any other exception raised for a record (e.g. a marshmallow ValidationError from an import deserializer) stops the whole harvest. Entries with transformation errors are not passed to the writers either, so the failure would never get to the harvest record.

create_datastream therefore wraps

  • each transformer in CatchingTransformer - any exception (and errors a transformer reports in errors without raising, as a TransformerError) is stored on the entry as exc (not in errors, so the datastream still hands the entry to the writers), the entry's data is replaced with {"record": None, "oai_record": <OAI record>} and the following transformers pass it through. The oai-service writer stores the exception on the harvest record and leaves the local record as it is;
  • each writer in CatchingWriter - any exception is stored on the entry as exc.

CatchingTransformer

Bases: BaseTransformer

Store any exception of the wrapped transformer on the entry instead of raising it.

Source code in oarepo_oaipmh_harvester/datastream.py 
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
class CatchingTransformer(BaseTransformer):
    """Store any exception of the wrapped transformer on the entry instead of raising it."""

    def __init__(self, transformer: BaseTransformer) -> None:
        """Wrap the transformer."""
        self._transformer = transformer

    @override
    def apply(self, stream_entry: StreamEntry, *args: Any, **kwargs: Any) -> StreamEntry:
        """Apply the wrapped transformer, unless a previous one has failed."""
        if stream_entry.exc is not None:
            return stream_entry
        oai_record = _oai_record(stream_entry.entry)
        try:
            result = self._transformer.apply(stream_entry, *args, **kwargs)
        except Exception as err:  # a single record must not stop the harvest
            current_app.logger.exception(
                "Error transforming %s", oai_record.header.identifier if oai_record else stream_entry.entry
            )
            return _failed(stream_entry, oai_record, err)
        if result.errors:
            # errors reported without raising - the datastream would not hand the entry to
            # the writers, so turn them into an exception that is stored on the harvest record
            message = "; ".join(str(error) for error in result.errors)
            return _failed(result, oai_record, TransformerError(f"{type(self._transformer).__name__}: {message}"))
        return result

__init__(transformer: BaseTransformer) -> None

Wrap the transformer.

Source code in oarepo_oaipmh_harvester/datastream.py 
67
68
69
def __init__(self, transformer: BaseTransformer) -> None:
    """Wrap the transformer."""
    self._transformer = transformer

apply(stream_entry: StreamEntry, *args: Any, **kwargs: Any) -> StreamEntry

Apply the wrapped transformer, unless a previous one has failed.

Source code in oarepo_oaipmh_harvester/datastream.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
@override
def apply(self, stream_entry: StreamEntry, *args: Any, **kwargs: Any) -> StreamEntry:
    """Apply the wrapped transformer, unless a previous one has failed."""
    if stream_entry.exc is not None:
        return stream_entry
    oai_record = _oai_record(stream_entry.entry)
    try:
        result = self._transformer.apply(stream_entry, *args, **kwargs)
    except Exception as err:  # a single record must not stop the harvest
        current_app.logger.exception(
            "Error transforming %s", oai_record.header.identifier if oai_record else stream_entry.entry
        )
        return _failed(stream_entry, oai_record, err)
    if result.errors:
        # errors reported without raising - the datastream would not hand the entry to
        # the writers, so turn them into an exception that is stored on the harvest record
        message = "; ".join(str(error) for error in result.errors)
        return _failed(result, oai_record, TransformerError(f"{type(self._transformer).__name__}: {message}"))
    return result

CatchingWriter

Bases: BaseWriter

Store any exception of the wrapped writer on the entry instead of raising it.

Source code in oarepo_oaipmh_harvester/datastream.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
class CatchingWriter(BaseWriter):
    """Store any exception of the wrapped writer on the entry instead of raising it."""

    def __init__(self, writer: BaseWriter) -> None:
        """Wrap the writer."""
        super().__init__()
        self._writer = writer
        self.is_async = getattr(writer, "is_async", False)

    @override
    def write(self, stream_entry: StreamEntry, *args: Any, **kwargs: Any) -> StreamEntry:
        """Write the entry with the wrapped writer."""
        try:
            return self._writer.write(stream_entry, *args, **kwargs)
        except Exception as err:  # a single record must not stop the harvest
            current_app.logger.exception("Error writing %s", stream_entry.entry)
            # we can't know the state of the db connection, rollback is a safe bet
            db.session.rollback()
            stream_entry.exc = err
            return stream_entry

    @override
    def write_many(self, stream_entries: list[StreamEntry], *args: Any, **kwargs: Any) -> list[StreamEntry]:
        """Write the entries one by one, so that a failure affects only its entry."""
        return [self.write(stream_entry, *args, **kwargs) for stream_entry in stream_entries]

__init__(writer: BaseWriter) -> None

Wrap the writer.

Source code in oarepo_oaipmh_harvester/datastream.py 
103
104
105
106
107
def __init__(self, writer: BaseWriter) -> None:
    """Wrap the writer."""
    super().__init__()
    self._writer = writer
    self.is_async = getattr(writer, "is_async", False)

write(stream_entry: StreamEntry, *args: Any, **kwargs: Any) -> StreamEntry

Write the entry with the wrapped writer.

Source code in oarepo_oaipmh_harvester/datastream.py 
109
110
111
112
113
114
115
116
117
118
119
@override
def write(self, stream_entry: StreamEntry, *args: Any, **kwargs: Any) -> StreamEntry:
    """Write the entry with the wrapped writer."""
    try:
        return self._writer.write(stream_entry, *args, **kwargs)
    except Exception as err:  # a single record must not stop the harvest
        current_app.logger.exception("Error writing %s", stream_entry.entry)
        # we can't know the state of the db connection, rollback is a safe bet
        db.session.rollback()
        stream_entry.exc = err
        return stream_entry

write_many(stream_entries: list[StreamEntry], *args: Any, **kwargs: Any) -> list[StreamEntry]

Write the entries one by one, so that a failure affects only its entry.

Source code in oarepo_oaipmh_harvester/datastream.py 
121
122
123
124
@override
def write_many(self, stream_entries: list[StreamEntry], *args: Any, **kwargs: Any) -> list[StreamEntry]:
    """Write the entries one by one, so that a failure affects only its entry."""
    return [self.write(stream_entry, *args, **kwargs) for stream_entry in stream_entries]

create_datastream(readers_config: list[dict], transformers_config: list[dict], writers_config: list[dict], **kwargs: Any) -> DataStream

Create a datastream (as DataStreamFactory.create) with error catching transformers and writers.

Source code in oarepo_oaipmh_harvester/datastream.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
def create_datastream(
    readers_config: list[dict],
    transformers_config: list[dict],
    writers_config: list[dict],
    **kwargs: Any,
) -> DataStream:
    """Create a datastream (as `DataStreamFactory.create`) with error catching transformers and writers."""
    return DataStream(
        readers=[ReaderFactory.create(config) for config in readers_config],
        transformers=[CatchingTransformer(TransformerFactory.create(config)) for config in transformers_config],
        writers=[CatchingWriter(WriterFactory.create(config)) for config in writers_config],
        **kwargs,
    )