codekingpro/portable-devtools

1import argparse2from contextlib import redirect_stderr, redirect_stdout3from typing import List4 5from httpie.context import Environment6from httpie.internal.update_warnings import _fetch_updates, _get_suppress_context7from httpie.status import ExitStatus8 9STATUS_FILE = '.httpie-test-daemon-status'10 11 12def _check_status(env):13    # This function is used only for the testing (test_update_warnings).14    # Since we don't want to trigger the fetch_updates (which would interact15    # with real world resources), we'll only trigger this pseudo task16    # and check whether the STATUS_FILE is created or not.17    import tempfile18    from pathlib import Path19 20    status_file = Path(tempfile.gettempdir()) / STATUS_FILE21    status_file.touch()22 23 24DAEMONIZED_TASKS = {25    'check_status': _check_status,26    'fetch_updates': _fetch_updates,27}28 29 30def _parse_options(args: List[str]) -> argparse.Namespace:31    parser = argparse.ArgumentParser()32    parser.add_argument('task_id')33    parser.add_argument('--daemon', action='store_true')34    return parser.parse_known_args(args)[0]35 36 37def is_daemon_mode(args: List[str]) -> bool:38    return '--daemon' in args39 40 41def run_daemon_task(env: Environment, args: List[str]) -> ExitStatus:42    options = _parse_options(args)43 44    assert options.daemon45    assert options.task_id in DAEMONIZED_TASKS46    with redirect_stdout(env.devnull), redirect_stderr(env.devnull):47        with _get_suppress_context(env):48            DAEMONIZED_TASKS[options.task_id](env)49 50    return ExitStatus.SUCCESS51