Coverage for src/oc_graphenricher/_storage.py: 96%

1# SPDX-FileCopyrightText: 2026 Arcangelo Massari <arcangelo.massari@unibo.it> 

2# 

3# SPDX-License-Identifier: ISC 

4 

5from __future__ import annotations 

6 

7import os 

8from typing import TYPE_CHECKING 

9 

10from oc_ocdm.storer import Storer 

11 

12from oc_graphenricher.storage import DirectoryStorage, Storage 

13 

14if TYPE_CHECKING: 

15 from oc_ocdm.graph.graph_set import GraphSet 

16 from oc_ocdm.prov.prov_set import ProvSet 

17 

18 

19def store_graph_set(graph_set: GraphSet, storage: Storage) -> None: 

20 storer = _storer(graph_set, storage) 

21 if isinstance(storage, DirectoryStorage): 

22 storer.store_all( 

23 _directory_output(storage.output_dir), 

24 graph_set.base_iri, 

25 storage.context_path, 

26 process_id=storage.process_id, 

27 ) 

28 else: 

29 storer.store_graphs_in_file(storage.graph_path, storage.context_path) 

30 

31 

32def store_provenance(provenance: ProvSet, storage: Storage) -> None: 

33 storer = _storer(provenance, storage) 

34 if isinstance(storage, DirectoryStorage): 

35 storer.store_all( 

36 _directory_output(storage.output_dir), 

37 provenance.base_iri, 

38 storage.context_path, 

39 process_id=storage.process_id, 

40 ) 

41 else: 

42 storer.store_graphs_in_file(storage.provenance_path, storage.context_path) 

43 

44 

45def _storer(entity_set: GraphSet | ProvSet, storage: Storage) -> Storer: 

46 if isinstance(storage, DirectoryStorage): 

47 return Storer( 

48 entity_set, 

49 repok=storage.repok, 

50 reperr=storage.reperr, 

51 context_map=storage.context_map, 

52 default_dir=storage.supplier_prefix, 

53 dir_split=storage.items_per_directory, 

54 n_file_item=storage.items_per_file, 

55 output_format=storage.output_format, 

56 zip_output=storage.zip_output, 

57 modified_entities=storage.modified_entities, 

58 ) 

59 return Storer( 

60 entity_set, 

61 repok=storage.repok, 

62 reperr=storage.reperr, 

63 context_map=storage.context_map, 

64 default_dir=storage.supplier_prefix, 

65 output_format=storage.output_format, 

66 zip_output=storage.zip_output, 

67 modified_entities=storage.modified_entities, 

68 ) 

69 

70 

71def _directory_output(directory: str) -> str: 

72 if directory.endswith(os.sep): 

73 return directory 

74 return directory + os.sep