test.py

def request_schema(schema, locations=None, put_into=None, example=None, add_to_refs=False, **kwargs):
    if callable(schema):
        schema = schema()
    # location kwarg added for compatibility with old versions
    locations = locations or []
    if not locations:
        locations = kwargs.pop("location", None)
        if locations:
            locations = [locations]
        else:
            locations = None

    options = {"required": kwargs.pop("required", False)}
    if locations:
        options["default_in"] = locations[0]

    def wrapper(func):
        if not hasattr(func, "__apispec__"):
            func.__apispec__ = {"schemas": [], "responses": {}, "parameters": []}
            func.__schemas__ = []

        _example = example or {}
        if _example:
            _example['add_to_refs'] = add_to_refs
        func.__apispec__["schemas"].append({"schema": schema, "options": options, "example": _example})
        # TODO: Remove this block?
        if locations and "body" in locations:
            body_schema_exists = (
                "body" in func_schema["locations"] for func_schema in func.__schemas__
            )
            if any(body_schema_exists):
                raise RuntimeError("Multiple body parameters are not allowed")

        func.__schemas__.append({"schema": schema, "locations": locations, "put_into": put_into})

        return func

    return wrapper

  
@request_schema(RequestSchema, example=example_for_request_schema)
async def handler_post_with_example_to_endpoint(request):
    return web.json_response({"msg": "done", "data": {}})

print('#' * 30)
print(handler_post_with_example_to_endpoint.__name__) # handler_post_with_example_to_endpoint
print(handler_post_with_example_to_endpoint.__apispec__["schemas"]) # [{'schema': <RequestSchema(many=False)>, 'options': {'required': False}, 'example': {'id': 1, 'name': 'test', 'bool_field': True, 'list_field': [1, 2, 3], 'add_to_refs': False}}]


@request_schema(RequestSchema, example=example_for_request_schema, add_to_refs=True)
async def handler_post_with_example_to_ref(request):
    return web.json_response({"msg": "done", "data": {}})

print('#' * 30)
print(handler_post_with_example_to_endpoint.__name__)  # handler_post_with_example_to_endpoint
print(handler_post_with_example_to_endpoint.__apispec__["schemas"]) # [{'schema': <RequestSchema(many=False)>, 'options': {'required': False}, 'example': {'id': 1, 'name': 'test', 'bool_field': True, 'list_field': [1, 2, 3], 'add_to_refs': True}}]
print('#' * 30)
print(handler_post_with_example_to_ref.__name__) # handler_post_with_example_to_ref
print(handler_post_with_example_to_ref.__apispec__["schemas"]) # [{'schema': <RequestSchema(many=False)>, 'options': {'required': False}, 'example': {'id': 1, 'name': 'test', 'bool_field': True, 'list_field': [1, 2, 3], 'add_to_refs': True}}]