Function wrapper

celery/app/base.py:158–199

(*task_args, **task_kwargs)

    @functools.wraps(task_fun)
    def wrapper(*task_args, **task_kwargs):
        # Validate task parameters if type hinted as BaseModel
        bound_args = task_signature.bind(*task_args, **task_kwargs)
        for arg_name, arg_value in bound_args.arguments.items():
            if type_hints and arg_name in type_hints:
                arg_annotation = type_hints[arg_name]
            else:
                arg_annotation = task_signature.parameters[arg_name].annotation

            optional_arg = get_optional_arg(arg_annotation)
            if optional_arg is not None and arg_value is not None:
                arg_annotation = optional_arg

            if annotation_issubclass(arg_annotation, BaseModel):
                bound_args.arguments[arg_name] = arg_annotation.model_validate(
                    arg_value,
                    strict=strict,
                    context={**context, 'celery_app': app, 'celery_task_name': task_name},
                )

        # Call the task with (potentially) converted arguments
        returned_value = task_fun(*bound_args.args, **bound_args.kwargs)

        # Dump Pydantic model if the returned value is an instance of pydantic.BaseModel *and* its
        # class matches the typehint
        if type_hints and 'return' in type_hints:
            return_annotation = type_hints['return']
        else:
            return_annotation = task_signature.return_annotation

        optional_return_annotation = get_optional_arg(return_annotation)
        if optional_return_annotation is not None:
            return_annotation = optional_return_annotation

        if (
            annotation_is_class(return_annotation)
            and isinstance(returned_value, BaseModel)
            and isinstance(returned_value, return_annotation)
        ):
            return returned_value.model_dump(**dump_kwargs)

        return returned_value

    return wrapper