Last active
March 21, 2023 22:55
-
-
Save wabiloo/386c7613da48a04228b0b8a60391c5b2 to your computer and use it in GitHub Desktop.
Python click experiments for REST API wrapper
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import click | |
class RestEndpointGroup(click.Group): | |
"""A click.Group sub-class that enables the use of command lines that | |
1. mirror REST endpoint structure | |
(eg. `mycli sources 123 slots 456` -> http://myapi/sources/:source_id/slots/:slot_id) | |
2. allow for implicit commands for `list` and `get` when no sub-commands are provided | |
on parent groups that support it. | |
(eg. `mycli sources` -> `mycli sources list`) | |
(eg. `mycli sources 123` -> `mycli sources 123 get`) | |
3. save automatically the ID to the context (for use deeper in the chain) | |
Inspired by https://stackoverflow.com/a/44056564/2215413""" | |
def parse_args(self, ctx, args): | |
# No sub-command? Then it's an implicit `list` | |
if len(args) == 0 and "list" in self.commands: | |
args.append("list") | |
# `list` command does not take an ID argument, | |
# so inject an empty one to prevent parse failure | |
if args[0] == "list": | |
args.insert(0, "") | |
# single argument, which is not a command? | |
# It must be an ID, and we treat it as an implicit `get` | |
if args[0] not in self.commands: | |
if (len(args) == 1) or (len(args) == 2 and args[1] in ["-h", "--help"]): | |
args.insert(1, "get") | |
# if the command is `list`, but preceded by a non-empty string, that's an error | |
if args[0] != "" and args[1] == "list": | |
raise click.BadArgumentUsage( | |
"A `list` command cannot be preceded by an object identifier" | |
) | |
# argument before command? It's an ID, | |
# and we save it automatically to the context object | |
if args[0] != "" and args[0] not in self.commands and args[1] in self.commands: | |
arg_name = self.params[0].name | |
ctx.obj[arg_name] = args[0] | |
super(RestEndpointGroup, self).parse_args(ctx, args) | |
@click.group() | |
@click.option("-d", "--dev", is_flag=True, required=False) | |
@click.option("-t", "--tenant", required=False) | |
@click.pass_context | |
def cli(ctx, dev, tenant): | |
BASE_URL = "http://mycli" | |
if dev: | |
BASE_URL = f"{BASE_URL}-DEV" | |
if tenant: | |
BASE_URL = f"{BASE_URL}/{tenant}" | |
ctx.obj = dict(url=BASE_URL) | |
@cli.group() | |
@click.pass_obj | |
def hello(obj): | |
pass | |
@cli.group(cls=RestEndpointGroup) | |
@click.argument("source_id", metavar="<source_id>") | |
@click.pass_obj | |
def sources(obj, source_id): | |
pass | |
@sources.command() | |
@click.option("--foo", required=False) | |
@click.pass_obj | |
def get(obj, foo): | |
url = f"{obj['url']}/sources/{obj['source_id']}/get" | |
if foo: | |
url = f"{url}?foo={foo}" | |
click.echo(url) | |
@sources.command() | |
@click.pass_obj | |
def list(obj): | |
click.echo(f"{obj['url']}/sources/list") | |
@sources.command() | |
@click.option("--baz", required=False) | |
@click.pass_obj | |
def read(obj, baz): | |
url = f"{obj['url']}/sources/{obj['source_id']}/read" | |
if baz: | |
url = f"{url}?baz={baz}" | |
click.echo(url) | |
@sources.group(cls=RestEndpointGroup) | |
@click.argument("slot_id", metavar="<slot_id>") | |
@click.pass_obj | |
def slots(obj, slot_id): | |
pass | |
@slots.command() | |
@click.pass_obj | |
def list(obj): | |
click.echo(f"{obj['url']}/sources/{obj['source_id']}/slots/list") | |
@slots.command() | |
@click.pass_obj | |
def get(obj): | |
url = f"{obj['url']}/sources/{obj['source_id']}/slots/{obj['slot_id']}/get" | |
click.echo(url) | |
@slots.command() | |
@click.option("--fake", required=False, default=False, is_flag=True) | |
@click.pass_obj | |
def remove(obj, fake): | |
url = f"{obj['url']}/sources/{obj['source_id']}/slots/{obj['slot_id']}/remove" | |
if fake: | |
url = f"{url}?fake=true" | |
click.echo(url) | |
if __name__ == "__main__": | |
cli() |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from click.testing import CliRunner | |
from rest_api_commands import cli | |
def _test_valid_command(command: str, endpoint: str): | |
runner = CliRunner() | |
result = runner.invoke(cli, command.split(" ")) | |
assert result.exit_code == 0 | |
assert result.output.startswith(endpoint) | |
def _test_valid_command_returns_help(command: str, endpoint: str): | |
runner = CliRunner() | |
result = runner.invoke(cli, command.split(" ")) | |
assert result.exit_code == 0 | |
assert result.output.startswith("Usage: ") | |
def _test_invalid_command_usage(command: str, endpoint: str): | |
runner = CliRunner() | |
result = runner.invoke(cli, command.split(" ")) | |
assert result.exit_code == 2 | |
assert result.output.startswith("Usage: cli") | |
def _test_invalid_command_error(command: str, endpoint: str): | |
runner = CliRunner() | |
result = runner.invoke(cli, command.split(" ")) | |
assert result.exit_code == 2 | |
assert result.output.startswith("Error: ") | |
def test_level1_no_id_explicit_list(): | |
_test_valid_command("sources list", "http://mycli/sources/list") | |
def test_level1_no_id_implicit_list(): | |
_test_valid_command("sources", "http://mycli/sources/list") | |
def test_level1_no_id_implicit_list_and_level0_flag(): | |
_test_valid_command("--dev sources", "http://mycli-DEV/sources/list") | |
def test_level1_spurious_id_for_list_command(): | |
_test_invalid_command_error("sources 123 list", "http://mycli-DEV/sources/list") | |
def test_level1_no_id_implicit_list_and_level0_option(): | |
_test_valid_command("--tenant 5 sources", "http://mycli/5/sources/list") | |
def test_level1_with_id_implicit_get(): | |
_test_valid_command("sources 123", "http://mycli/sources/123/get") | |
def test_level1_with_id_implicit_get_and_help(): | |
_test_valid_command_returns_help( | |
"sources 123 --help", "http://mycli/sources/123/get" | |
) | |
def test_level1_with_id_explicit_get(): | |
_test_valid_command("sources 123 get", "http://mycli/sources/123/get") | |
def test_level1_with_id_explicit_get_with_options(): | |
_test_valid_command( | |
"sources 123 get --foo bar", "http://mycli/sources/123/get?foo=bar" | |
) | |
def test_level1_with_id_command(): | |
_test_valid_command( | |
"sources 123 read --baz 5", "http://mycli/sources/123/read?baz=5" | |
) | |
def test_level1_missing_id_command(): | |
_test_invalid_command_usage( | |
"sources read --baz 5", "http://mycli/sources/123/read?baz=5" | |
) | |
def test_level2_no_id_implicit_list(): | |
_test_valid_command("sources 123 slots", "http://mycli/sources/123/slots/list") | |
def test_level2_no_id_explicit_list(): | |
_test_valid_command("sources 123 slots list", "http://mycli/sources/123/slots/list") | |
def test_level2_with_id_implicit_get(): | |
_test_valid_command( | |
"sources 123 slots 456", "http://mycli/sources/123/slots/456/get" | |
) | |
def test_level2_with_id_explicit_get(): | |
_test_valid_command( | |
"sources 123 slots 456 get", "http://mycli/sources/123/slots/456/get" | |
) | |
def test_level2_with_id_command(): | |
_test_valid_command( | |
"sources 123 slots 456 remove", "http://mycli/sources/123/slots/456/remove" | |
) | |
def test_level2_with_id_command_and_flag(): | |
_test_valid_command( | |
"sources 123 slots 456 remove --fake", | |
"http://mycli/sources/123/slots/456/remove?fake=true", | |
) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment