Skip to content

Instantly share code, notes, and snippets.

@wabiloo
Last active March 21, 2023 22:55
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save wabiloo/386c7613da48a04228b0b8a60391c5b2 to your computer and use it in GitHub Desktop.
Save wabiloo/386c7613da48a04228b0b8a60391c5b2 to your computer and use it in GitHub Desktop.
Python click experiments for REST API wrapper
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()
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