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 with this function to make them accessible remotely or create 'actions' out of them.
Parameters
----------
input_schema: JSON
schema for arguments to validate them.
output_schema: JSON
schema for return value, currently only used to inform clients which is supposed to validate on its won.
state: str | Tuple[str], optional
state machine state under which the object can executed. When not provided,
the action can be executed under any state.
**kwargs:
- safe: bool,
indicate in thing description if action is safe to execute
- idempotent: bool,
indicate in thing description if action is idempotent (for example, allows HTTP client to cache return value)
- synchronous: bool,
indicate in thing description if action is synchronous (not long running)
Returns
-------
Action
returns the callable object wrapped in an `Action` object
"""
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)):
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:
if global_config.validate_schemas:
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 global_config.validate_schemas and 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)))
if isinstance(input_schema, (BaseModel, RootModel)):
execution_info_validator.argument_schema = input_schema.model_json_schema()
elif isinstance(input_schema, dict):
execution_info_validator.argument_schema = input_schema
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 isinstance(output_schema, (BaseModel, RootModel)):
execution_info_validator.return_value_schema = output_schema.model_json_schema()
else:
try:
output_schema_model = get_return_type_from_signature(obj)
execution_info_validator.return_value_schema = output_schema_model.model_json_schema()
except Exception as ex:
warnings.warn(
f"Could not infer output schema for {obj.__name__} due to {ex}. " +
"Considering filing a bug report if you think this should have worked correctly",
category=RuntimeError
)
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