Skip to content

Instantly share code, notes, and snippets.

@FZambia
Last active October 21, 2019 06:47
Show Gist options
  • Star 17 You must be signed in to star a gist
  • Fork 5 You must be signed in to fork a gist
  • Save FZambia/5756470 to your computer and use it in GitHub Desktop.
Save FZambia/5756470 to your computer and use it in GitHub Desktop.
tornado's Subprocess class usage example. Minimal Tornado's version required - 3.1
from __future__ import print_function
from tornado.gen import Task, Return, coroutine
import tornado.process
from tornado.ioloop import IOLoop
import subprocess
import time
STREAM = tornado.process.Subprocess.STREAM
@coroutine
def call_subprocess(cmd, stdin_data=None, stdin_async=False):
"""
Wrapper around subprocess call using Tornado's Subprocess class.
"""
stdin = STREAM if stdin_async else subprocess.PIPE
sub_process = tornado.process.Subprocess(
cmd, stdin=stdin, stdout=STREAM, stderr=STREAM
)
if stdin_data:
if stdin_async:
yield Task(sub_process.stdin.write, stdin_data)
else:
sub_process.stdin.write(stdin_data)
if stdin_async or stdin_data:
sub_process.stdin.close()
result, error = yield [
Task(sub_process.stdout.read_until_close),
Task(sub_process.stderr.read_until_close)
]
raise Return((result, error))
def on_timeout():
print("timeout")
IOLoop.instance().stop()
@coroutine
def main():
seconds_to_wait = 3
deadline = time.time() + seconds_to_wait
# don't wait too long
IOLoop.instance().add_timeout(deadline, on_timeout)
# try to get output using synchronous PIPE for stdin
result, error = yield call_subprocess('wc', stdin_data="123")
print('stdin sync: ', result, error)
# try to get output using asynchronous STREAM for stdin
result, error = yield call_subprocess('wc', stdin_data="123", stdin_async=True)
print('stdin async: ', result, error)
IOLoop.instance().stop()
if __name__ == "__main__":
ioloop = IOLoop.instance()
ioloop.add_callback(main)
ioloop.start()
@JobsDong
Copy link

It's valuable to me. Thank you.

@JobsDong
Copy link

using shlex to format commands will be better.Just like shlex.split(cmd)

@Nilom
Copy link

Nilom commented Feb 18, 2014

this won't work when cmd comes with argument(s). So,

import shlex

.....

cmd = shlex.split(cmd)

will work.

@rms1000watt
Copy link

Thank you

@saintthor
Copy link

how to call this command?
curl -v -F 'token=zzz' -F 'file=@t.png' upload.target.com

@hellowrakesh123
Copy link

How can i get the exit code of the process? I tried using subprocess.returncode and subprocess.proc.returncode and it is set to None. For failure i would like to get the actual exit code.
Thanks in advance.!

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment