Coverage for /home/ubuntu/hidebound/python/hidebound/exporters/disk_exporter.py: 100%

1from typing import Any, Dict, List # noqa F401 

2 

3from pathlib import Path 

4import os 

5import shutil 

6 

7from schematics.types import StringType 

8 

9from hidebound.exporters.exporter_base import ExporterBase, ExporterConfigBase 

10import hidebound.core.tools as hbt 

11import hidebound.core.validators as vd 

12# ------------------------------------------------------------------------------ 

13 

14 

15class DiskConfig(ExporterConfigBase): 

16 ''' 

17 A class for validating configurations supplied to DiskExporter. 

18 

19 Attributes: 

20 name (str): Name of exporter. Must be 'disk'. 

21 target_directory (str): Target directory. 

22 ''' 

23 name = StringType( 

24 required=True, validators=[lambda x: vd.is_eq(x, 'disk')] 

25 ) # type: StringType 

26 target_directory = StringType( 

27 required=True, validators=[vd.is_legal_directory] 

28 ) # type: StringType 

29 

30 

31class DiskExporter(ExporterBase): 

32 @staticmethod 

33 def from_config(config): 

34 # type: (Dict) -> DiskExporter 

35 ''' 

36 Construct a DiskExporter from a given config. 

37 

38 Args: 

39 config (dict): Config dictionary. 

40 

41 Raises: 

42 DataError: If config is invalid. 

43 

44 Returns: 

45 DiskExporter: DiskExporter instance. 

46 ''' 

47 return DiskExporter(**config) 

48 

49 def __init__( 

50 self, 

51 target_directory, 

52 metadata_types=['asset', 'file', 'asset-chunk', 'file-chunk'], 

53 **kwargs, 

54 ): 

55 # type: (str, List[str], Any) -> None 

56 ''' 

57 Constructs a DiskExporter instance. 

58 Creates target directory if it does not exist. 

59 

60 Args: 

61 target_directory (str): Target directory. 

62 metadata_types (list, optional): List of metadata types for export. 

63 Default: [asset, file, asset-chunk, file-chunk]. 

64 

65 Raises: 

66 DataError: If config is invalid. 

67 ''' 

68 super().__init__(metadata_types=metadata_types) 

69 

70 config = dict( 

71 name='disk', 

72 target_directory=target_directory, 

73 metadata_types=metadata_types, 

74 ) 

75 DiskConfig(config).validate() 

76 # ---------------------------------------------------------------------- 

77 

78 self._target_directory = str(config['target_directory']) # type: str 

79 os.makedirs(self._target_directory, exist_ok=True) 

80 

81 def _export_content(self, metadata): 

82 # type: (Dict) -> None 

83 ''' 

84 Exports content from filepath in given metadata. 

85 

86 Args: 

87 metadata (dict): File metadata. 

88 ''' 

89 source = metadata['filepath'] 

90 target = Path( 

91 self._target_directory, 

92 'content', 

93 metadata['filepath_relative'], 

94 ) 

95 os.makedirs(Path(target).parent, exist_ok=True) 

96 shutil.copy(source, target) 

97 

98 def _export_asset(self, metadata): 

99 # type: (Dict) -> None 

100 ''' 

101 Exports metadata from single JSON file in hidebound/metadata/asset. 

102 

103 Args: 

104 metadata (dict): Asset metadata. 

105 ''' 

106 target = Path( 

107 self._target_directory, 

108 'metadata', 

109 'asset', 

110 metadata['asset_id'] + '.json', 

111 ) 

112 os.makedirs(Path(target).parent, exist_ok=True) 

113 hbt.write_json(metadata, target) 

114 

115 def _export_file(self, metadata): 

116 # type: (Dict) -> None 

117 ''' 

118 Exports metadata from single JSON file in hidebound/metadata/file. 

119 

120 Args: 

121 metadata (dict): File metadata. 

122 ''' 

123 target = Path( 

124 self._target_directory, 

125 'metadata', 

126 'file', 

127 metadata['file_id'] + '.json', 

128 ) 

129 os.makedirs(Path(target).parent, exist_ok=True) 

130 hbt.write_json(metadata, target) 

131 

132 def _export_asset_chunk(self, metadata): 

133 # type: (List[dict]) -> None 

134 ''' 

135 Exports content from single asset chunk in hidebound/metadata/asset-chunk. 

136 

137 Args: 

138 metadata (list[dict]): Asset metadata. 

139 ''' 

140 target = Path( 

141 self._target_directory, 

142 'metadata', 

143 'asset-chunk', 

144 f'hidebound-asset-chunk_{self._time}.json', 

145 ) 

146 os.makedirs(Path(target).parent, exist_ok=True) 

147 hbt.write_json(metadata, target) 

148 

149 def _export_file_chunk(self, metadata): 

150 # type: (List[dict]) -> None 

151 ''' 

152 Exports content from single file chunk in hidebound/metadata/file-chunk. 

153 

154 Args: 

155 metadata (list[dict]): File metadata. 

156 ''' 

157 target = Path( 

158 self._target_directory, 

159 'metadata', 

160 'file-chunk', 

161 f'hidebound-file-chunk_{self._time}.json', 

162 ) 

163 os.makedirs(Path(target).parent, exist_ok=True) 

164 hbt.write_json(metadata, target)