Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/nbconvert/writers/files.py: 22%

1"""Contains writer for writing nbconvert output to filesystem.""" 

2 

3# Copyright (c) IPython Development Team. 

4# Distributed under the terms of the Modified BSD License. 

5 

6import errno 

7import glob 

8import os 

9from pathlib import Path 

10 

11from traitlets import Unicode, observe 

12 

13from nbconvert.utils.io import link_or_copy 

14 

15from .base import WriterBase 

16 

17 

18class FilesWriter(WriterBase): 

19 """Consumes nbconvert output and produces files.""" 

20 

21 build_directory = Unicode( 

22 "", 

23 help="""Directory to write output(s) to. Defaults 

24 to output to the directory of each notebook. To recover 

25 previous default behaviour (outputting to the current 

26 working directory) use . as the flag value.""", 

27 ).tag(config=True) 

28 

29 relpath = Unicode( 

30 help="""When copying files that the notebook depends on, copy them in 

31 relation to this path, such that the destination filename will be 

32 os.path.relpath(filename, relpath). If FilesWriter is operating on a 

33 notebook that already exists elsewhere on disk, then the default will be 

34 the directory containing that notebook.""" 

35 ).tag(config=True) 

36 

37 # Make sure that the output directory exists. 

38 @observe("build_directory") 

39 def _build_directory_changed(self, change): 

40 new = change["new"] 

41 if new: 

42 self._makedir(new) 

43 

44 def __init__(self, **kw): 

45 """Initialize the writer.""" 

46 super().__init__(**kw) 

47 self._build_directory_changed({"new": self.build_directory}) 

48 

49 def _makedir(self, path, mode=0o755): 

50 """ensure that a directory exists 

51 

52 If it doesn't exist, try to create it and protect against a race condition 

53 if another process is doing the same. 

54 

55 The default permissions are 755, which differ from os.makedirs default of 777. 

56 """ 

57 if not os.path.exists(path): 

58 self.log.info("Making directory %s", path) 

59 try: 

60 os.makedirs(path, mode=mode) 

61 except OSError as e: 

62 if e.errno != errno.EEXIST: 

63 raise 

64 elif not os.path.isdir(path): 

65 raise OSError("%r exists but is not a directory" % path) 

66 

67 def _write_items(self, items, build_dir): 

68 """Write a dict containing filename->binary data""" 

69 for filename, data in items: 

70 # Determine where to write the file to 

71 dest = os.path.join(build_dir, filename) 

72 path = os.path.dirname(dest) 

73 self._makedir(path) 

74 

75 # Write file 

76 self.log.debug("Writing %i bytes to %s", len(data), dest) 

77 with open(dest, "wb") as f: 

78 f.write(data) 

79 

80 def write(self, output, resources, notebook_name=None, **kw): 

81 """ 

82 Consume and write Jinja output to the file system. Output directory 

83 is set via the 'build_directory' variable of this instance (a 

84 configurable). 

85 

86 See base for more... 

87 """ 

88 

89 # Verify that a notebook name is provided. 

90 if notebook_name is None: 

91 msg = "notebook_name" 

92 raise TypeError(msg) 

93 

94 # Pull the extension and subdir from the resources dict. 

95 output_extension = resources.get("output_extension", None) 

96 

97 # Get the relative path for copying files 

98 resource_path = resources.get("metadata", {}).get("path", "") 

99 relpath = self.relpath or resource_path 

100 build_directory = self.build_directory or resource_path 

101 

102 # Write the extracted outputs to the destination directory. 

103 # NOTE: WE WRITE EVERYTHING AS-IF IT'S BINARY. THE EXTRACT FIG 

104 # PREPROCESSOR SHOULD HANDLE UNIX/WINDOWS LINE ENDINGS... 

105 

106 items = resources.get("outputs", {}).items() 

107 if items: 

108 self.log.info( 

109 "Support files will be in %s", 

110 os.path.join(resources.get("output_files_dir", ""), ""), 

111 ) 

112 self._write_items(items, build_directory) 

113 

114 # Write the extracted attachments 

115 # if ExtractAttachmentsOutput specified a separate directory 

116 attachs = resources.get("attachments", {}).items() 

117 if attachs: 

118 self.log.info( 

119 "Attachments will be in %s", 

120 os.path.join(resources.get("attachment_files_dir", ""), ""), 

121 ) 

122 self._write_items(attachs, build_directory) 

123 

124 # Copy referenced files to output directory 

125 if build_directory: 

126 for filename in self.files: 

127 # Copy files that match search pattern 

128 for matching_filename in glob.glob(filename): 

129 # compute the relative path for the filename 

130 if relpath != "": # noqa 

131 dest_filename = os.path.relpath(matching_filename, relpath) 

132 else: 

133 dest_filename = matching_filename 

134 

135 # Make sure folder exists. 

136 dest = os.path.join(build_directory, dest_filename) 

137 path = os.path.dirname(dest) 

138 self._makedir(path) 

139 

140 # Copy if destination is different. 

141 if os.path.normpath(dest) != os.path.normpath(matching_filename): 

142 self.log.info("Copying %s -> %s", matching_filename, dest) 

143 link_or_copy(matching_filename, dest) 

144 

145 # Determine where to write conversion results. 

146 dest = notebook_name + output_extension if output_extension is not None else notebook_name 

147 dest = Path(build_directory) / dest 

148 

149 # Write conversion results. 

150 self.log.info("Writing %i bytes to %s", len(output), dest) 

151 if isinstance(output, str): 

152 with open(dest, "w", encoding="utf-8") as f: 

153 f.write(output) 

154 else: 

155 with open(dest, "wb") as f: 

156 f.write(output) 

157 

158 return dest