Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/graphviz/piping.py: 55%

1"""Pipe DOT code objects through Graphviz ``dot``.""" 

2 

3import codecs 

4import logging 

5import typing 

6 

7from . import _tools 

8from . import backend 

9from . import exceptions 

10from . import base 

11from . import encoding 

12 

13__all__ = ['Pipe'] 

14 

15 

16log = logging.getLogger(__name__) 

17 

18 

19class Pipe(encoding.Encoding, base.Base, backend.Pipe): 

20 """Pipe source lines through the Graphviz layout command.""" 

21 

22 @typing.overload 

23 def pipe(self, 

24 format: typing.Optional[str] = ..., 

25 renderer: typing.Optional[str] = ..., 

26 formatter: typing.Optional[str] = ..., 

27 neato_no_op: typing.Union[bool, int, None] = ..., 

28 quiet: bool = ..., *, 

29 engine: typing.Optional[str] = ..., 

30 encoding: None = ...) -> bytes: 

31 """Return bytes with default ``encoding=None``.""" 

32 

33 @typing.overload 

34 def pipe(self, 

35 format: typing.Optional[str] = ..., 

36 renderer: typing.Optional[str] = ..., 

37 formatter: typing.Optional[str] = ..., 

38 neato_no_op: typing.Union[bool, int, None] = ..., 

39 quiet: bool = ..., *, 

40 engine: typing.Optional[str] = ..., 

41 encoding: str) -> str: 

42 """Return string when given encoding.""" 

43 

44 @typing.overload 

45 def pipe(self, 

46 format: typing.Optional[str] = ..., 

47 renderer: typing.Optional[str] = ..., 

48 formatter: typing.Optional[str] = ..., 

49 neato_no_op: typing.Union[bool, int, None] = ..., 

50 quiet: bool = ..., *, 

51 engine: typing.Optional[str] = ..., 

52 encoding: typing.Optional[str]) -> typing.Union[bytes, str]: 

53 """Return bytes or string depending on encoding argument.""" 

54 

55 def pipe(self, 

56 format: typing.Optional[str] = None, 

57 renderer: typing.Optional[str] = None, 

58 formatter: typing.Optional[str] = None, 

59 neato_no_op: typing.Union[bool, int, None] = None, 

60 quiet: bool = False, *, 

61 engine: typing.Optional[str] = None, 

62 encoding: typing.Optional[str] = None) -> typing.Union[bytes, str]: 

63 """Return the source piped through the Graphviz layout command. 

64 

65 Args: 

66 format: The output format used for rendering 

67 (``'pdf'``, ``'png'``, etc.). 

68 renderer: The output renderer used for rendering 

69 (``'cairo'``, ``'gd'``, ...). 

70 formatter: The output formatter used for rendering 

71 (``'cairo'``, ``'gd'``, ...). 

72 neato_no_op: Neato layout engine no-op flag. 

73 quiet (bool): Suppress ``stderr`` output 

74 from the layout subprocess. 

75 engine: Layout engine for rendering 

76 (``'dot'``, ``'neato'``, ...). 

77 encoding: Encoding for decoding the stdout. 

78 

79 Returns: 

80 Bytes or if encoding is given decoded string 

81 (stdout of the layout command). 

82 

83 Raises: 

84 ValueError: If ``engine``, ``format``, ``renderer``, or ``formatter`` 

85 are unknown. 

86 graphviz.RequiredArgumentError: If ``formatter`` is given 

87 but ``renderer`` is None. 

88 graphviz.ExecutableNotFound: If the Graphviz ``dot`` executable 

89 is not found. 

90 graphviz.CalledProcessError: If the returncode (exit status) 

91 of the rendering ``dot`` subprocess is non-zero. 

92 

93 Example: 

94 >>> doctest_mark_exe() 

95 >>> import graphviz 

96 >>> source = 'graph { spam }' 

97 >>> graphviz.Source(source, format='svg').pipe()[:14] 

98 b'<?xml version=' 

99 >>> graphviz.Source(source, format='svg').pipe(encoding='ascii')[:14] 

100 '<?xml version=' 

101 >>> graphviz.Source(source, format='svg').pipe(encoding='utf-8')[:14] 

102 '<?xml version=' 

103 """ 

104 return self._pipe_legacy(format, 

105 renderer=renderer, 

106 formatter=formatter, 

107 neato_no_op=neato_no_op, 

108 quiet=quiet, 

109 engine=engine, 

110 encoding=encoding) 

111 

112 @_tools.deprecate_positional_args(supported_number=1, ignore_arg='self') 

113 def _pipe_legacy(self, 

114 format: typing.Optional[str] = None, 

115 renderer: typing.Optional[str] = None, 

116 formatter: typing.Optional[str] = None, 

117 neato_no_op: typing.Union[bool, int, None] = None, 

118 quiet: bool = False, *, 

119 engine: typing.Optional[str] = None, 

120 encoding: typing.Optional[str] = None) -> typing.Union[bytes, str]: 

121 return self._pipe_future(format, 

122 renderer=renderer, 

123 formatter=formatter, 

124 neato_no_op=neato_no_op, 

125 quiet=quiet, 

126 engine=engine, 

127 encoding=encoding) 

128 

129 def _pipe_future(self, format: typing.Optional[str] = None, *, 

130 renderer: typing.Optional[str] = None, 

131 formatter: typing.Optional[str] = None, 

132 neato_no_op: typing.Union[bool, int, None] = None, 

133 quiet: bool = False, 

134 engine: typing.Optional[str] = None, 

135 encoding: typing.Optional[str] = None) -> typing.Union[bytes, str]: 

136 args, kwargs = self._get_pipe_parameters(engine=engine, 

137 format=format, 

138 renderer=renderer, 

139 formatter=formatter, 

140 neato_no_op=neato_no_op, 

141 quiet=quiet, 

142 verify=True) 

143 

144 args.append(iter(self)) 

145 

146 if encoding is not None: 

147 if codecs.lookup(encoding) is codecs.lookup(self.encoding): 

148 # common case: both stdin and stdout need the same encoding 

149 return self._pipe_lines_string(*args, encoding=encoding, **kwargs) 

150 try: 

151 raw = self._pipe_lines(*args, input_encoding=self.encoding, **kwargs) 

152 except exceptions.CalledProcessError as e: 

153 *args, output, stderr = e.args 

154 if output is not None: 

155 output = output.decode(self.encoding) 

156 if stderr is not None: 

157 stderr = stderr.decode(self.encoding) 

158 raise e.__class__(*args, output=output, stderr=stderr) 

159 else: 

160 return raw.decode(encoding) 

161 return self._pipe_lines(*args, input_encoding=self.encoding, **kwargs)