-
Notifications
You must be signed in to change notification settings - Fork 6
ExperimentalAPI CLI #428
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
ExperimentalAPI CLI #428
Changes from all commits
987cc17
de74ef7
a56e831
e0d9734
ea1014d
fbe190a
518b608
ef71903
3e6fb5c
cccccd3
75e97f1
88cabec
17201fd
85705e4
008ef12
9b4472f
37d43b6
4d0b7ae
e9f4e28
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,16 +1,47 @@ | ||
| import json | ||
| import logging | ||
| import sys | ||
| from datetime import date, datetime | ||
| from decimal import Decimal | ||
| from enum import Enum | ||
| from functools import wraps | ||
| from typing import Union | ||
| from importlib.metadata import version as importlib_version | ||
| from typing import Any, Union | ||
| from uuid import UUID | ||
|
|
||
| import typer | ||
| from groundlight_openapi_client.model_utils import OpenApiModel | ||
| from pydantic import BaseModel | ||
| from typing_extensions import get_origin | ||
|
|
||
| from groundlight import Groundlight | ||
| from groundlight import ExperimentalApi, Groundlight | ||
| from groundlight.client import ApiTokenError | ||
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
| cli_app = typer.Typer( | ||
| context_settings={"help_option_names": ["-h", "--help"], "max_content_width": 800}, | ||
| ) | ||
|
|
||
|
|
||
| @cli_app.callback(invoke_without_command=True) | ||
| def _main( | ||
| ctx: typer.Context, | ||
| version: bool = typer.Option(False, "--version", "-v", is_eager=True, help="Show the SDK version and exit."), | ||
| ): | ||
| if version: | ||
| print(importlib_version("groundlight")) | ||
| raise typer.Exit() | ||
| if ctx.invoked_subcommand is None: | ||
| print(ctx.get_help()) | ||
|
|
||
|
|
||
| experimental_app = typer.Typer( | ||
| no_args_is_help=True, | ||
| help="Experimental commands — may change or be removed without notice.", | ||
| context_settings={"help_option_names": ["-h", "--help"], "max_content_width": 800}, | ||
| ) | ||
| cli_app.add_typer(experimental_app, name="exp", rich_help_panel="Subcommands") | ||
|
|
||
|
|
||
| def is_cli_supported_type(annotation): | ||
|
|
@@ -21,15 +52,66 @@ def is_cli_supported_type(annotation): | |
| return annotation in (int, float, bool) | ||
|
|
||
|
|
||
| def class_func_to_cli(method): | ||
| def is_cli_representable(annotation) -> bool: | ||
| """Returns True if the annotation is a type Typer can natively represent as a CLI argument. | ||
|
|
||
| Primitive scalar types, Enum subclasses, and Union types (handled separately) are considered | ||
| representable. Complex types like dict, list, bytes, and custom model classes are not. | ||
| """ | ||
| if annotation in (str, int, float, bool): | ||
| return True | ||
| if isinstance(annotation, type) and issubclass(annotation, Enum): | ||
| return True | ||
| if get_origin(annotation) is Union: | ||
| return True | ||
| return False | ||
|
|
||
|
|
||
| def _json_default(obj: Any) -> Any: | ||
| """Fallback serializer for json.dumps for types the stdlib encoder doesn't handle. | ||
|
|
||
| Covers common types that appear in OpenAPI client to_dict() output. Unknown types | ||
| fall back to str() rather than raising, so CLI output is always usable. | ||
| """ | ||
| if isinstance(obj, (datetime, date)): | ||
| return obj.isoformat() | ||
| if isinstance(obj, Decimal): | ||
| return float(obj) | ||
| if isinstance(obj, UUID): | ||
| return str(obj) | ||
| if isinstance(obj, Enum): | ||
| return obj.value | ||
| return str(obj) | ||
|
|
||
|
|
||
| def _format_result(result: Any) -> str: | ||
| """Format a CLI result value as a human-readable string. | ||
|
|
||
| Pydantic models and OpenAPI client objects are serialized to indented JSON. | ||
| Plain dicts and lists are also JSON. Everything else falls back to str(). | ||
| """ | ||
| if isinstance(result, BaseModel): | ||
| return result.model_dump_json(indent=2) | ||
| if isinstance(result, OpenApiModel): | ||
| return json.dumps(result.to_dict(), indent=2, default=_json_default) | ||
| if isinstance(result, (dict, list)): | ||
| return json.dumps(result, indent=2, default=_json_default) | ||
| return str(result) | ||
|
|
||
|
|
||
| def class_func_to_cli(method, is_experimental: bool = False): | ||
| """ | ||
| Given the class method, create a method with the identical signature to provide the help documentation and | ||
| but only instantiates the class when the method is actually called. | ||
| Given a class method, return a wrapper function with the same signature that Typer can | ||
| register as a CLI command. The wrapper instantiates ExperimentalApi at call time (which | ||
| also provides all stable Groundlight methods via inheritance), so a single instantiation | ||
| path serves both stable and experimental commands. | ||
|
|
||
| If is_experimental is True, a warning is printed to stderr before the method runs. | ||
| """ | ||
|
|
||
| # We create a fake class and fake method so we have the correct annotations for typer to use | ||
| # When we wrap the fake method, we only use the fake method's name to access the real method | ||
| # and attach it to a Groundlight instance that we create at function call time | ||
| # We create a fake class and fake method so we have the correct annotations for typer to use. | ||
| # When we wrap the fake method, we only use the fake method's name to look up and call the | ||
| # real method on an ExperimentalApi instance created at call time. | ||
| class FakeClass: | ||
| pass | ||
|
|
||
|
|
@@ -38,14 +120,26 @@ class FakeClass: | |
|
|
||
| @wraps(fake_method) | ||
| def wrapper(*args, **kwargs): | ||
| gl = Groundlight() | ||
| gl_method = vars(Groundlight)[fake_method.__name__] | ||
| gl_bound_method = gl_method.__get__(gl, Groundlight) # pylint: disable=all | ||
| print(gl_bound_method(*args, **kwargs)) # this is where we output to the console | ||
| if is_experimental: | ||
| print( | ||
| f"Warning: '{fake_method.__name__}' is an experimental command and may change without notice.", | ||
| file=sys.stderr, | ||
| ) | ||
| gl = ExperimentalApi() | ||
| # Typer sees the fake method's annotations (for correct CLI argument types), but the | ||
| # actual call goes to the real method on a live ExperimentalApi instance. The fake | ||
| # method's name is identical to the real one, so getattr resolves to the correct | ||
| # implementation, including inherited Groundlight methods. | ||
| bound_method = getattr(gl, fake_method.__name__) | ||
| result = bound_method(*args, **kwargs) | ||
| if result is not None: | ||
| print(_format_result(result)) | ||
|
|
||
| # not recommended practice to directly change annotations, but gets around Typer not supporting Union types | ||
| cli_unsupported_params = [] | ||
| for name, annotation in method.__annotations__.items(): | ||
| if name == "return": | ||
| continue | ||
| if get_origin(annotation) is Union: | ||
| # If we can submit a string, we take the string from the cli | ||
| if str in annotation.__args__: | ||
|
|
@@ -60,6 +154,11 @@ def wrapper(*args, **kwargs): | |
| break | ||
| if not found_supported_type: | ||
| cli_unsupported_params.append(name) | ||
| elif is_experimental and not is_cli_representable(annotation): | ||
| # For experimental methods only: proactively flag non-Union types that Typer cannot | ||
| # represent (e.g. dict, list, custom models) so the caller can skip them gracefully | ||
| # before Typer raises a deferred RuntimeError at cli_app() invocation time. | ||
| cli_unsupported_params.append(name) | ||
| # Ideally we could just not list the unsupported params, but it doesn't seem natively supported by Typer | ||
| # and requires more metaprogamming than makes sense at the moment. For now, we require methods to support str | ||
| for param in cli_unsupported_params: | ||
|
|
@@ -71,13 +170,102 @@ def wrapper(*args, **kwargs): | |
| return wrapper | ||
|
|
||
|
|
||
| # Desired display order of command groups in the CLI help output. | ||
| _GROUP_ORDER = [ | ||
| "Account", | ||
| "Detectors", | ||
| "Image Queries", | ||
| "ML Pipelines & Priming", | ||
| "Notes", | ||
| "Utilities", | ||
| ] | ||
|
|
||
| # Maps method names to their rich_help_panel group label for the CLI help output. | ||
| # Applies to both stable and experimental commands. | ||
| _COMMAND_GROUPS: dict[str, str] = { | ||
|
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I think this adds maintenance overhead, new methods need to belong to a group. The original design tried to avoid that additional overhead, but in the age of AI maybe we don't care
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I added a test to ensure that each method has a CLI group. It will be an extra step, but a worthwhile extra step that cannot be forgotten. |
||
| # Account | ||
| "whoami": "Account", | ||
| "get_month_to_date_usage": "Account", | ||
| # Detectors | ||
| "get_detector": "Detectors", | ||
| "get_detector_by_name": "Detectors", | ||
| "list_detectors": "Detectors", | ||
| "create_detector": "Detectors", | ||
| "get_or_create_detector": "Detectors", | ||
| "delete_detector": "Detectors", | ||
| "create_binary_detector": "Detectors", | ||
| "create_counting_detector": "Detectors", | ||
| "create_multiclass_detector": "Detectors", | ||
| "create_bounding_box_detector": "Detectors", | ||
| "create_detector_group": "Detectors", | ||
| "list_detector_groups": "Detectors", | ||
| "create_roi": "Detectors", | ||
| "update_detector_confidence_threshold": "Detectors", | ||
| "update_detector_status": "Detectors", | ||
| "update_detector_escalation_type": "Detectors", | ||
| "reset_detector": "Detectors", | ||
| "update_detector_name": "Detectors", | ||
| "create_text_recognition_detector": "Detectors", | ||
| "get_detector_evaluation": "Detectors", | ||
| "get_detector_metrics": "Detectors", | ||
| "download_mlbinary": "Detectors", | ||
| # Image Queries | ||
| "get_image_query": "Image Queries", | ||
| "list_image_queries": "Image Queries", | ||
| "submit_image_query": "Image Queries", | ||
| "ask_confident": "Image Queries", | ||
| "ask_ml": "Image Queries", | ||
| "ask_async": "Image Queries", | ||
| "wait_for_confident_result": "Image Queries", | ||
| "wait_for_ml_result": "Image Queries", | ||
| "get_image": "Image Queries", | ||
| "add_label": "Image Queries", | ||
| # Notes | ||
| "get_notes": "Notes", | ||
| "create_note": "Notes", | ||
| # ML Pipelines & Priming | ||
| "list_detector_pipelines": "ML Pipelines & Priming", | ||
| "list_priming_groups": "ML Pipelines & Priming", | ||
| "create_priming_group": "ML Pipelines & Priming", | ||
| "get_priming_group": "ML Pipelines & Priming", | ||
| "delete_priming_group": "ML Pipelines & Priming", | ||
| # Utilities | ||
| "edge_base_url": "Utilities", | ||
| "get_raw_headers": "Utilities", | ||
| } | ||
|
|
||
|
|
||
| def _cli_sort_key(item: tuple) -> tuple: | ||
| """Sort key for CLI command registration that controls group and within-group ordering. | ||
|
|
||
| Commands are ordered first by their group's position in _GROUP_ORDER, then alphabetically | ||
| by method name within each group. | ||
| """ | ||
| name, _ = item | ||
| group = _COMMAND_GROUPS.get(name) | ||
| group_rank = _GROUP_ORDER.index(group) if group in _GROUP_ORDER else len(_GROUP_ORDER) | ||
| return (group_rank, name) | ||
|
|
||
|
|
||
| def groundlight(): | ||
| """Entry point for the groundlight CLI.""" | ||
| try: | ||
| # For each method in the Groundlight class, create a function that can be called from the command line | ||
| for name, method in vars(Groundlight).items(): | ||
| stable_names = {n for n, m in vars(Groundlight).items() if callable(m) and not n.startswith("_")} | ||
|
|
||
| for name, method in sorted(vars(Groundlight).items(), key=_cli_sort_key): | ||
| if callable(method) and not name.startswith("_"): | ||
| cli_func = class_func_to_cli(method) | ||
| cli_app.command()(cli_func) | ||
| cli_app.command(rich_help_panel=_COMMAND_GROUPS[name])(cli_func) | ||
|
|
||
| for name, method in sorted(vars(ExperimentalApi).items(), key=_cli_sort_key): | ||
| if not callable(method) or name.startswith("_") or name in stable_names: | ||
| continue | ||
| try: | ||
| cli_func = class_func_to_cli(method, is_experimental=True) | ||
| experimental_app.command(rich_help_panel=_COMMAND_GROUPS[name])(cli_func) | ||
| except Exception as e: # pylint: disable=broad-except | ||
| logger.debug("Skipping experimental CLI command '%s': %s", name, e) | ||
|
|
||
| cli_app() | ||
| except ApiTokenError as e: | ||
| print(e) | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Heavily biased as the one who wrote the original, but there should be more comments around this one line. This is the magic line that makes the whole thing happen
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done. I added a comment.