Source code for datumaro.components.contexts.importer
# Copyright (C) 2019-2022 Intel Corporation
#
# SPDX-License-Identifier: MIT
from typing import NoReturn, Tuple
import attr
from attr import define, field
from datumaro.components.errors import AnnotationImportError, DatumaroError, ItemImportError
from datumaro.components.progress_reporting import NullProgressReporter, ProgressReporter
class _ImportFail(DatumaroError):
pass
[docs]
class ImportErrorPolicy:
[docs]
def report_item_error(self, error: Exception, *, item_id: Tuple[str, str]) -> None:
"""
Allows to report a problem with a dataset item.
If this function returns, the extractor must skip the item.
"""
if not isinstance(error, _ImportFail):
ie = ItemImportError(item_id)
ie.__cause__ = error
return self._handle_item_error(ie)
else:
raise error
[docs]
def report_annotation_error(self, error: Exception, *, item_id: Tuple[str, str]) -> None:
"""
Allows to report a problem with a dataset item annotation.
If this function returns, the extractor must skip the annotation.
"""
if not isinstance(error, _ImportFail):
ie = AnnotationImportError(item_id)
ie.__cause__ = error
return self._handle_annotation_error(ie)
else:
raise error
def _handle_item_error(self, error: ItemImportError) -> None:
"""This function must either call fail() or return."""
self.fail(error)
def _handle_annotation_error(self, error: AnnotationImportError) -> None:
"""This function must either call fail() or return."""
self.fail(error)
[docs]
def fail(self, error: Exception) -> NoReturn:
raise _ImportFail from error
[docs]
class FailingImportErrorPolicy(ImportErrorPolicy):
pass
[docs]
@define(eq=False)
class ImportContext:
progress_reporter: ProgressReporter = field(
default=None, converter=attr.converters.default_if_none(factory=NullProgressReporter)
)
error_policy: ImportErrorPolicy = field(
default=None, converter=attr.converters.default_if_none(factory=FailingImportErrorPolicy)
)
[docs]
class NullImportContext(ImportContext):
pass