import inspect
import sys
from collections.abc import Callable, Coroutine
from functools import partial
from typing import TYPE_CHECKING, Any, Literal, TypeVar, overload
from cyclopts._result_action import ResultAction
if sys.version_info < (3, 11): # pragma: no cover
from typing_extensions import assert_never
else: # pragma: no cover
from typing import assert_never
if TYPE_CHECKING:
from cyclopts.core import App
V = TypeVar("V")
# App will be lazily imported to avoid circular imports
App = None # type: ignore[assignment]
def _run_maybe_async_command(
command: Callable,
bound: inspect.BoundArguments | None = None,
backend: Literal["asyncio", "trio"] = "asyncio",
):
"""Run a command, handling both sync and async cases.
If the command is async, an async context will be created to run it.
Parameters
----------
command : Callable
The command to execute.
bound : inspect.BoundArguments | None
Bound arguments for the command. If None, command is called with no arguments.
backend : Literal["asyncio", "trio"]
The async backend to use if the command is async.
Returns
-------
return_value: Any
The value the command function returns.
"""
if not inspect.iscoroutinefunction(command):
if bound is None:
return command()
else:
return command(*bound.args, **bound.kwargs)
if backend == "asyncio":
import asyncio
if bound is None:
return asyncio.run(command())
else:
return asyncio.run(command(*bound.args, **bound.kwargs))
elif backend == "trio":
import trio
if bound is None:
return trio.run(command)
else:
return trio.run(partial(command, *bound.args, **bound.kwargs))
else: # pragma: no cover
assert_never(backend)
@overload
def run(callable: Callable[..., Coroutine[None, None, V]], /, *, result_action: Literal["return_value"]) -> V: ...
@overload
def run(callable: Callable[..., V], /, *, result_action: Literal["return_value"]) -> V: ...
@overload
def run(
callable: Callable[..., Coroutine[None, None, Any]], /, *, result_action: ResultAction | None = None
) -> Any: ...
@overload
def run(callable: Callable[..., Any], /, *, result_action: ResultAction | None = None) -> Any: ...
def run(callable, /, *, result_action: ResultAction | None = None):
"""Run the given callable as a CLI command.
The callable may also be a coroutine function.
This function is syntax sugar for very simple use cases, and is roughly equivalent to:
.. code-block:: python
from cyclopts import App
app = App()
app.default(callable)
app()
Parameters
----------
callable
The function to execute as a CLI command.
result_action
How to handle the command's return value. If not specified, uses the default
``"print_non_int_sys_exit"`` which calls :func:`sys.exit` with the appropriate code.
Can be set to ``"return_value"`` to return the result directly for testing/embedding.
Example usage:
.. code-block:: python
import cyclopts
def main(name: str, age: int):
print(f"Hello {name}, you are {age} years old.")
cyclopts.run(main)
"""
global App
if App is None:
from cyclopts.core import App as _App
App = _App
app = App(result_action=result_action)
app.default(callable)
return app()
