Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 25 additions & 0 deletions httpie/cli/definition.py
Original file line number Diff line number Diff line change
Expand Up @@ -239,6 +239,31 @@

""",
)
processing_options.add_argument(
'--sequence',
action='store_true',
default=False,
short_help='Execute multiple requests from STDIN sequentially.',
help="""
Read and execute multiple HTTPie request definitions from STDIN,
one request per line, sequentially.

Each line should contain a complete HTTPie command without the
leading "http" program name, e.g.:

GET https://httpbin.org/get
POST https://httpbin.org/post name=alice
GET https://httpbin.org/headers

Example usage:

$ cat requests.http | http --sequence
$ echo -e "GET httpbin.org/get\nPOST httpbin.org/post foo=bar" | http --sequence

Requests are executed strictly in order, one at a time.

""",
)


#######################################################################
Expand Down
91 changes: 88 additions & 3 deletions httpie/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -126,9 +126,9 @@ def handle_generic_error(e, annotation=None):
original_exc = unwrap_context(exc)
if isinstance(original_exc, socket.gaierror):
if original_exc.errno == socket.EAI_AGAIN:
annotation = '\nCouldnt connect to a DNS server. Please check your connection and try again.'
annotation = '\nCouldn\'t connect to a DNS server. Please check your connection and try again.'
elif original_exc.errno == socket.EAI_NONAME:
annotation = '\nCouldnt resolve the given hostname. Please check the URL and try again.'
annotation = '\nCouldn\'t resolve the given hostname. Please check the URL and try again.'
propagated_exc = original_exc
else:
propagated_exc = exc
Expand Down Expand Up @@ -167,10 +167,95 @@ def main(
)


def run_sequence_mode(args: argparse.Namespace, env: Environment) -> ExitStatus:
"""
Execute multiple HTTPie request definitions from STDIN sequentially.

Each line in STDIN is treated as a separate HTTPie command without
the leading 'http' program name.

Example input:
GET https://httpbin.org/get
POST https://httpbin.org/post name=alice
GET https://httpbin.org/headers
"""
import shlex
from .cli.definition import parser

exit_status = ExitStatus.SUCCESS
request_lines = []

# Read all request definitions from STDIN
try:
for line in env.stdin:
line = line.strip()
if line and not line.startswith('#'):
request_lines.append(line)
except KeyboardInterrupt:
env.stderr.write('\n')
return ExitStatus.ERROR_CTRL_C

if not request_lines:
env.log_error('No request definitions found in STDIN.')
return ExitStatus.ERROR

# Execute each request sequentially
for i, request_line in enumerate(request_lines, 1):
if env.stdout_isatty:
env.stderr.write(f'\n[{i}/{len(request_lines)}] {request_line}\n')

try:
# Parse the request line as if it were CLI arguments
# Add --ignore-stdin to prevent argparser from using stdin for request body
request_args = shlex.split(request_line) + ['--ignore-stdin']

# Create a fresh environment for this request to avoid stdin conflicts
request_env = Environment(
stdin=None, # No stdin for individual requests in sequence mode
stdout=env.stdout,
stderr=env.stderr,
)
request_env.program_name = env.program_name

# Parse the request
request_namespace = parser.parse_args(
args=request_args,
env=request_env,
)

# Execute the request (call _program to avoid sequence check loop)
request_exit_status = _program(request_namespace, request_env)

# Track the worst exit status
if request_exit_status != ExitStatus.SUCCESS:
exit_status = request_exit_status

except SystemExit as e:
# Handle SystemExit from argument parser errors
if e.code != ExitStatus.SUCCESS:
exit_status = ExitStatus.ERROR if e.code is None or e.code != 0 else ExitStatus(e.code)
except Exception as e:
env.log_error(f'Error executing request [{i}]: {e}')
exit_status = ExitStatus.ERROR

return exit_status


def program(args: argparse.Namespace, env: Environment) -> ExitStatus:
"""
The main program without error handling.

"""
# Handle --sequence mode: execute multiple requests from STDIN
if getattr(args, 'sequence', False):
return run_sequence_mode(args, env)

return _program(args, env)


def _program(args: argparse.Namespace, env: Environment) -> ExitStatus:
"""
The actual program implementation without the --sequence check.
"""
# TODO: Refactor and drastically simplify, especially so that the separator logic is elsewhere.
exit_status = ExitStatus.SUCCESS
Expand Down Expand Up @@ -209,7 +294,7 @@ def request_body_read_callback(chunk: bytes):
force_separator = False
prev_with_body = False

# Process messages as theyre generated
# Process messages as they're generated
for message in messages:
output_options = OutputOptions.from_message(message, args.output_options)

Expand Down
149 changes: 149 additions & 0 deletions tests/test_sequence_mode.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,149 @@
"""Tests for --sequence mode: execute multiple requests from STDIN."""
import pytest
from io import StringIO
from unittest.mock import patch, MagicMock

from httpie.core import run_sequence_mode, program
from httpie.context import Environment
from httpie.status import ExitStatus


class TestSequenceMode:
"""Test the --sequence feature for executing multiple requests from STDIN."""

def test_sequence_empty_stdin(self):
"""Sequence mode with empty STDIN should return error."""
env = Environment()
env.stdin = StringIO('')

args = MagicMock()
args.sequence = True

result = run_sequence_mode(args, env)

assert result == ExitStatus.ERROR

def test_sequence_only_comments(self):
"""Sequence mode with only comments should return error."""
env = Environment()
env.stdin = StringIO('# This is a comment\n# Another comment\n')

args = MagicMock()
args.sequence = True

result = run_sequence_mode(args, env)

assert result == ExitStatus.ERROR

def test_sequence_parses_request_lines(self):
"""Sequence mode should parse request lines from STDIN."""
env = Environment()
env.stdin = StringIO('GET http://example.com\nPOST http://example.com data=test\n')
env.stdout = StringIO()
env.stdout_isatty = False

args = MagicMock()
args.sequence = True

# Mock _program to avoid actual HTTP requests
with patch('httpie.core._program') as mock_program:
mock_program.return_value = ExitStatus.SUCCESS

result = run_sequence_mode(args, env)

# Should have called _program twice (once per request)
assert mock_program.call_count == 2
assert result == ExitStatus.SUCCESS

def test_sequence_skips_empty_lines(self):
"""Sequence mode should skip empty lines."""
env = Environment()
env.stdin = StringIO('\nGET http://example.com\n\nPOST http://example.com\n\n')
env.stdout = StringIO()
env.stdout_isatty = False

args = MagicMock()
args.sequence = True

with patch('httpie.core._program') as mock_program:
mock_program.return_value = ExitStatus.SUCCESS

result = run_sequence_mode(args, env)

# Should have called _program twice (empty lines skipped)
assert mock_program.call_count == 2

def test_sequence_handles_keyboard_interrupt(self):
"""Sequence mode should handle KeyboardInterrupt gracefully."""
env = Environment()

# Simulate KeyboardInterrupt during stdin read
class InterruptingStringIO(StringIO):
def __iter__(self):
return self
def __next__(self):
raise KeyboardInterrupt()

env.stdin = InterruptingStringIO()
env.stdout = StringIO()

args = MagicMock()
args.sequence = True

result = run_sequence_mode(args, env)

assert result == ExitStatus.ERROR_CTRL_C

def test_sequence_propagates_error_status(self):
"""Sequence mode should return error if any request fails."""
env = Environment()
env.stdin = StringIO('GET http://example.com\nPOST http://example.com\n')
env.stdout = StringIO()
env.stdout_isatty = False

args = MagicMock()
args.sequence = True

with patch('httpie.core._program') as mock_program:
# First call succeeds, second fails
mock_program.side_effect = [ExitStatus.SUCCESS, ExitStatus.ERROR]

result = run_sequence_mode(args, env)

assert result == ExitStatus.ERROR

def test_sequence_handles_request_exception(self):
"""Sequence mode should handle exceptions during request execution."""
env = Environment()
env.stdin = StringIO('GET http://example.com\n')
env.stdout = StringIO()
env.stdout_isatty = False

args = MagicMock()
args.sequence = True

with patch('httpie.core._program') as mock_program:
mock_program.side_effect = Exception('Request failed')

result = run_sequence_mode(args, env)

assert result == ExitStatus.ERROR

def test_sequence_executes_multiple_requests(self):
"""Sequence mode should execute multiple requests sequentially."""
env = Environment()
env.stdin = StringIO('GET http://example.com\nPOST http://example.com\nPUT http://example.com\n')
env.stdout = StringIO()
env.stdout_isatty = False

args = MagicMock()
args.sequence = True

with patch('httpie.core._program') as mock_program:
mock_program.return_value = ExitStatus.SUCCESS

result = run_sequence_mode(args, env)

# Should have called _program 3 times for 3 requests
assert mock_program.call_count == 3
assert result == ExitStatus.SUCCESS
Loading