def action(
input_schema: JSON | BaseModel | RootModel | None = None,
output_schema: JSON | BaseModel | RootModel | None = None,
state: str | Enum | None = None,
**kwargs,
) -> 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.
"""
def inner(obj):
input_schema = inner._arguments.get("input_schema", None)
output_schema = inner._arguments.get("output_schema", None)
state = inner._arguments.get("state", None)
kwargs = inner._arguments.get("kwargs", {})
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.execution_info.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__}")
execution_info_validator = ActionInfoValidator()
if state is not None:
if isinstance(state, (Enum, str)):
execution_info_validator.state = (state,)
else:
execution_info_validator.state = state
if "request" in getfullargspec(obj).kwonlyargs:
execution_info_validator.request_as_argument = True
execution_info_validator.isaction = True
execution_info_validator.obj = original
execution_info_validator.create_task = kwargs.get("create_task", False)
execution_info_validator.safe = kwargs.get("safe", False)
execution_info_validator.idempotent = kwargs.get("idempotent", False)
execution_info_validator.synchronous = kwargs.get("synchronous", True)
if isclassmethod(original):
execution_info_validator.iscoroutine = has_async_def(obj)
execution_info_validator.isclassmethod = True
elif issubklass(obj, ParameterizedFunction):
execution_info_validator.iscoroutine = iscoroutinefunction(obj.__call__)
execution_info_validator.isparameterized = True
else:
execution_info_validator.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 - {str(ex)}. "
+ "Considering filing a bug report if you think this should have worked correctly",
category=RuntimeWarning,
)
if input_schema:
if isinstance(input_schema, dict):
execution_info_validator.schema_validator = JSONSchemaValidator(input_schema)
elif issubklass(input_schema, (BaseModel, RootModel)):
execution_info_validator.schema_validator = PydanticSchemaValidator(input_schema)
else:
raise TypeError(
"input schema must be a JSON schema or a Pydantic model, got {}".format(type(input_schema))
)
execution_info_validator.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 {str(ex)}. "
+ "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 isinstance(output_schema, dict):
jsonschema.Draft7Validator.check_schema(output_schema)
execution_info_validator.return_value_schema = output_schema
elif issubklass(output_schema, (BaseModel, RootModel)):
execution_info_validator.return_value_schema = output_schema
else:
raise TypeError(
"output schema must be a JSON schema or a Pydantic model, got {}".format(type(output_schema))
)
final_obj = Action(original) # type: Action
final_obj.execution_info = execution_info_validator
return final_obj
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.keys()):
raise ValueError(
"Only 'safe', 'idempotent', 'synchronous' are allowed as keyword arguments, "
+ f"unknown arguments found {kwargs.keys()}"
)
inner._arguments = dict(
input_schema=input_schema,
output_schema=output_schema,
state=state,
kwargs=kwargs,
)
return inner