def action(
input_schema: JSON | BaseModel | RootModel | None = None,
output_schema: JSON | BaseModel | RootModel | None = None,
state: str | Enum | None = None,
**kwargs,
) -> Callable[[Any], Action]:
"""
Decorate on your methods to make them accessible remotely or create 'actions' out of them.
When used with hardware, actions generally command the hardware to do something.
Parameters
----------
input_schema: JSON | BaseModel | RootModel, optional
schema for arguments to validate
output_schema: JSON | BaseModel | RootModel, optional
schema for return value, currently only used to inform clients which are supposed to validate on their own
state: str | Tuple[str], optional
state machine state under which the action can be executed. When not provided, the action can be executed
under any state.
**kwargs:
additional keyword arguments to specify action characteristics:
- `synchronous`: bool,
indicate in thing description if action is synchronous (not long running/threaded or async) - completes
in a deterministic (& usually) short period of time, default `True`
- `threaded`: bool,
indicate that a method/action should be run in a separate thread, default `False`.
Alternative to `synchronous` for non-async methods.
- `create_task`: bool,
indicate that a method/action should be run in a new task, default `True`.
Alternative to `synchronous` for async methods.
- `safe`: bool,
indicate in thing description if action is safe to execute, default `False`
- `idempotent`: bool,
indicate in thing description if action is idempotent (for example, allows HTTP clients to cache return value),
default `False`
Returns
-------
Action
returns the callable object wrapped in an `Action` object. When accessed at instance level,
a `BoundSyncAction` or `BoundAsyncAction` object is returned.
Raises
------
TypeError
if the decorated object is not a function or method, or if the input/output schema is of invalid type
ValueError
if the decorated function is a dunder method, or if unknown keyword arguments are provided
"""
def inner(obj):
input_schema = inner._arguments.get("input_schema", None) # ty: ignore[unresolved-attribute]
output_schema = inner._arguments.get("output_schema", None) # ty: ignore[unresolved-attribute]
state = inner._arguments.get("state", None) # ty: ignore[unresolved-attribute]
kwargs = inner._arguments.get("kwargs", {}) # ty: ignore[unresolved-attribute]
original = obj
if (
not isinstance(obj, (FunctionType, MethodType, Action, BoundAction))
and not isclassmethod(obj)
and not issubklass(obj, ParameterizedFunction)
):
raise TypeError(f"target for action or is not a function/method. Given type {type(obj)}") from None
if isclassmethod(obj):
obj = obj.__func__
if isinstance(obj, (Action, BoundAction)):
if (obj if isinstance(obj, Action) else obj.action).isclassmethod:
raise RuntimeError("cannot wrap a classmethod as action once again, please skip")
warnings.warn(
f"{obj.name} is already wrapped as an action, wrapping it again with newer settings.",
category=UserWarning,
)
obj = obj.obj
if obj.__name__.startswith("__"):
raise ValueError(f"dunder objects cannot become remote : {obj.__name__}")
action = Action(original) # type: Action
action.state = Tuple(
default=None,
item_type=(Enum, str),
allow_None=True,
accept_list=True,
accept_item=True,
).validate_and_adapt(state)
if "request" in getfullargspec(obj).kwonlyargs:
action.request_as_argument = True
action.create_task = kwargs.get("create_task", False)
action.safe = kwargs.get("safe", False)
action.idempotent = kwargs.get("idempotent", False)
action.synchronous = kwargs.get("synchronous", True)
if isclassmethod(original):
action.iscoroutine = has_async_def(obj)
action.isclassmethod = True
elif issubklass(obj, ParameterizedFunction):
action.iscoroutine = iscoroutinefunction(obj.__call__)
action.isparameterized = True
else:
action.iscoroutine = iscoroutinefunction(obj)
if not input_schema:
try:
input_schema = get_input_model_from_signature(obj, remove_first_positional_arg=True)
except Exception as ex:
warnings.warn(
f"Could not infer input schema for {obj.__name__} due to - {ex!s}. "
+ "Considering filing a bug report if you think this should have worked correctly",
category=RuntimeWarning,
)
if input_schema:
if not SchemaValidators.is_supported(input_schema):
raise TypeError(
"no registered schema validator can validate against the input schema "
+ f"of {obj.__name__}, which is of type {type(input_schema)}. Supply a JSON schema, "
+ "a pydantic model, or register a validator that matches it with "
+ "SchemaValidators.register()."
)
if global_config.VALIDATE_SCHEMAS:
SchemaValidators.check_schema(input_schema)
action.argument_schema = input_schema
if not output_schema:
try:
output_schema = get_return_type_from_signature(obj)
except Exception as ex:
warnings.warn(
f"Could not infer output schema for {obj.__name__} due to {ex!s}. "
+ "Considering filing a bug report if you think this should have worked correctly",
category=RuntimeWarning,
)
if output_schema:
# output is not validated by us, so we just check the schema and dont create a validator
if not SchemaValidators.is_supported(output_schema):
raise TypeError(
"no registered schema validator can validate against the output schema "
+ f"of {obj.__name__}, which is of type {type(output_schema)}. Supply a JSON schema, "
+ "a pydantic model, or register a validator that matches it with "
+ "SchemaValidators.register()."
)
if global_config.VALIDATE_SCHEMAS:
SchemaValidators.check_schema(output_schema)
action.return_value_schema = output_schema
return action
if callable(input_schema):
raise TypeError(
"input schema should be a JSON or pydantic BaseModel, not a function/method, "
+ "did you decorate your action wrongly? use @action() instead of @action"
)
if any(key not in __action_kw_arguments__ for key in kwargs):
raise ValueError(
"Only 'safe', 'idempotent', 'synchronous' are allowed as keyword arguments, "
+ f"unknown arguments found {kwargs.keys()}"
)
inner._arguments = dict( # ty: ignore[unresolved-attribute]
input_schema=input_schema,
output_schema=output_schema,
state=state,
kwargs=kwargs,
)
return inner