File: sphinx_pipeinclude.py

package info (click to toggle)
s3ql 5.2.3%2Bdfsg-2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 4,416 kB
  • sloc: python: 19,027; cpp: 408; makefile: 147; sh: 133
file content (59 lines) | stat: -rw-r--r-- 1,705 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
'''
sphinx_pipe.py - this file is part of S3QL.

Implements a Sphinx extension that provides a `pipeinclude` directive
to include the output of a program.


Copyright © 2008 Nikolaus Rath <Nikolaus@rath.org>

This work can be distributed under the terms of the GNU GPLv3.
'''

import os.path
import shlex
import subprocess
import sys
import tempfile

from docutils.parsers.rst.directives.misc import Include
from docutils.utils.error_reporting import SafeString


class PipeInclude(Include):
    """
    Include program output as ReST source.
    """

    def run(self):
        # To maximize code reuse, we just write the output in a temporary
        # file and call the base class. Otherwise we'd have to copy & paste
        # all the code to handle start-line, end-line etc options.

        source = self.state_machine.input_lines.source(
            self.lineno - self.state_machine.input_offset - 1
        )
        source_dir = os.path.dirname(os.path.abspath(source))

        command = self.arguments[0]
        command_list = shlex.split(command)

        if command_list[0] == 'python':
            command_list[0] = sys.executable

        with tempfile.NamedTemporaryFile() as fh:
            exitcode = subprocess.call(command_list, stdout=fh, cwd=source_dir)
            if exitcode != 0:
                raise self.severe(
                    'Problems with "%s" directive:\n'
                    'Command %s returned with exit code %d'
                    % (self.name, SafeString(command), exitcode)
                )

            fh.flush()
            self.arguments[0] = fh.name
            return super().run()


def setup(app):
    app.add_directive('pipeinclude', PipeInclude)