Skip to content

Utils

DatasetError

Bases: Exception

Custom exception to report different error types to Scicat

Source code in backend/archiver/flows/utils.py 
11
12
13
14
class DatasetError(Exception):
    """Custom exception to report different error types to Scicat
    """
    pass

StoragePaths

Helper class to create paths in scratch and LTS folder

Source code in backend/archiver/flows/utils.py 
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
class StoragePaths:
    """Helper class to create paths in scratch and LTS folder
    """

    @ staticmethod
    def scratch_folder(dataset_id: str) -> Path:
        return StoragePaths.scratch_archival_root() / StoragePaths._relative_dataset_folder(dataset_id)

    @ staticmethod
    def scratch_archival_root() -> Path:
        return Variables().ARCHIVER_SCRATCH_FOLDER / "archival"

    @ staticmethod
    def _relative_dataset_folder(dataset_id: str) -> Path:
        return Path("openem-network") / "datasets" / dataset_id

    _relative_datablocks_folder: Path = Path("datablocks")
    _relative_raw_files_folder: Path = Path("raw_files")

    @ staticmethod
    def relative_datablocks_folder(dataset_id: str):
        return StoragePaths._relative_dataset_folder(dataset_id) / StoragePaths._relative_datablocks_folder

    @ staticmethod
    def relative_raw_files_folder(dataset_id: str):
        return StoragePaths._relative_dataset_folder(dataset_id) / StoragePaths._relative_raw_files_folder

    @ staticmethod
    def scratch_archival_datablocks_folder(dataset_id: str) -> Path:
        return StoragePaths.scratch_archival_root() / StoragePaths.relative_datablocks_folder(dataset_id)

    @ staticmethod
    def scratch_archival_raw_files_folder(dataset_id: str) -> Path:
        return StoragePaths.scratch_archival_root() / StoragePaths.relative_raw_files_folder(dataset_id)

    @ staticmethod
    def lts_datablocks_folder(dataset_id: str) -> Path:
        return Variables().LTS_STORAGE_ROOT / StoragePaths.relative_datablocks_folder(dataset_id)

SystemError

Bases: Exception

Custom exception to report different error types to Scicat

Source code in backend/archiver/flows/utils.py 
17
18
19
20
class SystemError(Exception):
    """Custom exception to report different error types to Scicat
    """
    pass

report_archival_error(dataset_id, state, task_run, token)

Report an error of an archival job of a dataset. Differntiates betwen "DatasetError" (User error, e.g. missing files) and SystemError (transient error).

Parameters:

Name Type Description Default
dataset_id str

dataset id

 required
state State

task run state

 required
task_run TaskRun

task run

 required
Source code in backend/archiver/flows/utils.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def report_archival_error(dataset_id: str, state: State, task_run: TaskRun, token: SecretStr):
    """Report an error of an archival job of a dataset. Differntiates betwen "DatasetError" (User error, e.g. missing files)
    and SystemError (transient error).

    Args:
        dataset_id (str): dataset id
        state (State): task run state
        task_run (TaskRun): task run
    """
    try:
        state.result()
    except DatasetError:
        report_dataset_user_error(dataset_id=dataset_id, token=token)
    except SystemError:
        report_dataset_system_error(dataset_id=dataset_id, token=token)
    except Exception:
        # TODO: add some info about unknown errors
        report_dataset_system_error(dataset_id=dataset_id, token=token)

report_retrieval_error(dataset_id, state, task_run, token)

Report a retrieval error of a job of a dataset. Differntiates betwen "DatasetError" (User error, e.g. missing files) and SystemError (transient error).

Parameters:

Name Type Description Default
dataset_id str

dataset id

 required
state State

task run state

 required
task_run TaskRun

task run

 required
Source code in backend/archiver/flows/utils.py 
43
44
45
46
47
48
49
50
51
52
53
def report_retrieval_error(dataset_id: str, state: State, task_run: TaskRun, token: SecretStr):
    """Report a retrieval error of a job of a dataset. Differntiates betwen "DatasetError" (User error, e.g. missing files)
    and SystemError (transient error).

    Args:
        dataset_id (str): dataset id
        state (State): task run state
        task_run (TaskRun): task run
    """

    report_dataset_retrieval_error(dataset_id=dataset_id, token=token)