|
1 | 1 | from inspect import isasyncgenfunction, iscoroutinefunction, isfunction, isgeneratorfunction |
2 | | -from typing import Final |
| 2 | +from typing import Any, Callable, Final |
| 3 | + |
| 4 | +import click |
3 | 5 |
|
4 | 6 | MARKER_MAGIC: Final[str] = "07c9b9cc" |
5 | 7 | MARKER_DRIVERCALL: Final[str] = "marker_drivercall" |
6 | 8 | MARKER_STREAMCALL: Final[str] = "marker_streamcall" |
7 | 9 | MARKER_STREAMING_DRIVERCALL: Final[str] = "marker_streamingdrivercall" |
| 10 | +MARKER_HELP: Final[str] = "marker_help" |
8 | 11 |
|
9 | 12 |
|
10 | | -def export(func): |
11 | | - """ |
12 | | - Decorator for exporting method as driver call |
13 | | - """ |
14 | | - if isasyncgenfunction(func) or isgeneratorfunction(func): |
15 | | - setattr(func, MARKER_STREAMING_DRIVERCALL, MARKER_MAGIC) |
16 | | - elif iscoroutinefunction(func) or isfunction(func): |
17 | | - setattr(func, MARKER_DRIVERCALL, MARKER_MAGIC) |
18 | | - else: |
19 | | - raise ValueError(f"unsupported exported function {func}") |
20 | | - return func |
| 13 | +def export(func: Callable | None = None, *, help: str | None = None) -> Callable: |
| 14 | + """Decorator for exporting method as driver call""" |
| 15 | + def decorator(f: Callable) -> Callable: |
| 16 | + if isasyncgenfunction(f) or isgeneratorfunction(f): |
| 17 | + setattr(f, MARKER_STREAMING_DRIVERCALL, MARKER_MAGIC) |
| 18 | + elif iscoroutinefunction(f) or isfunction(f): |
| 19 | + setattr(f, MARKER_DRIVERCALL, MARKER_MAGIC) |
| 20 | + else: |
| 21 | + raise ValueError(f"unsupported exported function {f}") |
| 22 | + if help is not None: |
| 23 | + setattr(f, MARKER_HELP, help) |
| 24 | + return f |
| 25 | + return decorator(func) if func else decorator |
| 26 | + |
21 | 27 |
|
| 28 | +def exportstream(func: Callable | None = None, *, help: str | None = None) -> Callable: |
| 29 | + """Decorator for exporting method as stream""" |
| 30 | + def decorator(f: Callable) -> Callable: |
| 31 | + setattr(f, MARKER_STREAMCALL, MARKER_MAGIC) |
| 32 | + if help is not None: |
| 33 | + setattr(f, MARKER_HELP, help) |
| 34 | + return f |
| 35 | + return decorator(func) if func else decorator |
22 | 36 |
|
23 | | -def exportstream(func): |
| 37 | + |
| 38 | +class DriverClickGroup(click.Group): |
24 | 39 | """ |
25 | | - Decorator for exporting method as stream |
| 40 | + Custom Click Group to use methods_description for command help. |
| 41 | +
|
| 42 | + Usage: |
| 43 | + def cli(self): |
| 44 | + base = DriverClickGroup(self, help="Default driver description") |
| 45 | +
|
| 46 | + @base.command() # Automatically uses string from methods_description['on'] |
| 47 | + def on(): |
| 48 | + self.on() |
| 49 | +
|
| 50 | + return base |
26 | 51 | """ |
27 | | - setattr(func, MARKER_STREAMCALL, MARKER_MAGIC) |
28 | | - return func |
| 52 | + |
| 53 | + def __init__(self, client: Any, *args: Any, **kwargs: Any) -> None: |
| 54 | + """ |
| 55 | + Initialize a DriverClickGroup bound to a client. |
| 56 | +
|
| 57 | + :param client: DriverClient instance (provides methods_description) |
| 58 | + :param args: Arguments passed to click.Group |
| 59 | + :param kwargs: Keyword arguments passed to click.Group |
| 60 | + """ |
| 61 | + super().__init__(*args, **kwargs) |
| 62 | + self.client = client |
| 63 | + |
| 64 | + def command(self, *args: Any, **kwargs: Any) -> Callable: |
| 65 | + """ |
| 66 | + Decorator to register a client-side command with automatic help text. |
| 67 | +
|
| 68 | + Priority order for help text: |
| 69 | + 1. methods_description from server (highest priority) |
| 70 | + 2. help= parameter explicitly provided in @base.command(help="...") |
| 71 | + 3. Empty string (fallback) |
| 72 | + """ |
| 73 | + def decorator(f: Callable) -> click.Command: |
| 74 | + # Determine command name (from kwargs or function name) |
| 75 | + name = kwargs.get('name') or f.__name__ |
| 76 | + |
| 77 | + # Priority order for help text: |
| 78 | + # 1. methods_description from server |
| 79 | + # 2. help= parameter explicitly provided by client |
| 80 | + # 3. Empty string |
| 81 | + if name in self.client.methods_description: |
| 82 | + kwargs['help'] = self.client.methods_description[name] |
| 83 | + elif 'help' not in kwargs: |
| 84 | + kwargs['help'] = '' |
| 85 | + |
| 86 | + return super(DriverClickGroup, self).command(*args, **kwargs)(f) |
| 87 | + |
| 88 | + return decorator |
0 commit comments