codekingpro/portable-devtools

1import argparse2 3from httpie.sessions import SESSIONS_DIR_NAME, get_httpie_session4from httpie.status import ExitStatus5from httpie.context import Environment6from httpie.legacy import v3_1_0_session_cookie_format, v3_2_0_session_header_format7from httpie.manager.cli import missing_subcommand, parser8from httpie.utils import is_version_greater9 10 11FIXERS_TO_VERSIONS = {12    '3.1.0': v3_1_0_session_cookie_format.fix_layout,13    '3.2.0': v3_2_0_session_header_format.fix_layout,14}15 16 17def cli_sessions(env: Environment, args: argparse.Namespace) -> ExitStatus:18    action = args.cli_sessions_action19    if action is None:20        parser.error(missing_subcommand('cli', 'sessions'))21 22    if action == 'upgrade':23        return cli_upgrade_session(env, args)24    elif action == 'upgrade-all':25        return cli_upgrade_all_sessions(env, args)26    else:27        raise ValueError(f'Unexpected action: {action}')28 29 30def upgrade_session(env: Environment, args: argparse.Namespace, hostname: str, session_name: str):31    session = get_httpie_session(32        env=env,33        config_dir=env.config.directory,34        session_name=session_name,35        host=hostname,36        url=hostname,37        suppress_legacy_warnings=True38    )39 40    session_name = session.path.stem41    if session.is_new():42        env.log_error(f'{session_name!r} @ {hostname!r} does not exist.')43        return ExitStatus.ERROR44 45    fixers = [46        fixer47        for version, fixer in FIXERS_TO_VERSIONS.items()48        if is_version_greater(version, session.version)49    ]50 51    if len(fixers) == 0:52        env.stdout.write(f'{session_name!r} @ {hostname!r} is already up to date.\n')53        return ExitStatus.SUCCESS54 55    for fixer in fixers:56        fixer(session, hostname, args)57 58    session.save(bump_version=True)59    env.stdout.write(f'Upgraded {session_name!r} @ {hostname!r} to v{session.version}\n')60    return ExitStatus.SUCCESS61 62 63def cli_upgrade_session(env: Environment, args: argparse.Namespace) -> ExitStatus:64    return upgrade_session(65        env,66        args=args,67        hostname=args.hostname,68        session_name=args.session69    )70 71 72def cli_upgrade_all_sessions(env: Environment, args: argparse.Namespace) -> ExitStatus:73    session_dir_path = env.config_dir / SESSIONS_DIR_NAME74 75    status = ExitStatus.SUCCESS76    for host_path in session_dir_path.iterdir():77        hostname = host_path.name78        for session_path in host_path.glob("*.json"):79            session_name = session_path.stem80            status |= upgrade_session(81                env,82                args=args,83                hostname=hostname,84                session_name=session_name85            )86    return status87