code: def requests(
      func: Optional[
          Callable[
              [
                  "DocumentArray",
                  Dict,
                  "DocumentArray",
                  List["DocumentArray"],
                  List["DocumentArray"],
              ],
              Optional[Union["DocumentArray", Dict]],
          ]
      ] = None,
      *,
      on: Optional[Union[str, Sequence[str]]] = None,
      request_schema: Optional[Type[DocumentArray]] = None,
      response_schema: Optional[
          Union[Type[DocumentArray], Type[Document]]
      ] = None,
  ):
      """
      `@requests` defines the endpoints of an Executor. It has a keyword `on=` to
      define the endpoint.
  
      A class method decorated with plain `@requests` (without `on=`) is the default
      handler for all endpoints.
      That means, it is the fallback handler for endpoints that are not found.
  
      EXAMPLE USAGE
  
      .. code-block:: python
  
          from jina import Executor, requests, Flow
          from docarray import Document
  
          # define Executor with custom `@requests` endpoints
          class MyExecutor(Executor):
              @requests(on='/index')
              def index(self, docs, **kwargs):
                  print(docs)  # index docs here
  
              @requests(on=['/search', '/query'])
              def search(self, docs, **kwargs):
                  print(docs)  # perform search here
  
              @requests  # default/fallback endpoint
              def foo(self, docs, **kwargs):
                  print(docs)  # process docs here
  
  
          f = Flow().add(uses=MyExecutor)  # add your Executor to a Flow
          with f:
              f.post(
                  on='/index', inputs=Document(text='I am here!')
              )  # send doc to `index` method
              f.post(
                  on='/search', inputs=Document(text='Who is there?')
              )  # send doc to `search` method
              f.post(
                  on='/query', inputs=Document(text='Who is there?')
              )  # send doc to `search` method
              f.post(on='/bar', inputs=Document(text='Who is there?'))  # send doc to
              # `foo` method
  
      :param func: the method to decorate
      :param on: the endpoint string, by convention starts with `/`
      :param request_schema: the type of the input document
      :param response_schema: the type of the output document
      :return: decorated function
      """
      from jina.constants import (
          __args_executor_func__,
          __default_endpoint__,
      )
  
      if func:
          setattr(func, "__is_generator__", is_generator(func))
  
      class FunctionMapper:
          def __init__(self, fn):
              if fn:
                  setattr(fn, "__is_generator__", is_generator(fn))
              self._batching_decorator = None
              self._write_decorator = None
              fn = self._unwrap_batching_decorator(fn)
              fn = self._unwrap_write_decorator(fn)
              arg_spec = inspect.getfullargspec(fn)
              if (
                  not arg_spec.varkw
                  and not __args_executor_func__.issubset(arg_spec.args)
              ):
                  raise TypeError(
                      f"{fn} accepts only {arg_spec.args} which is fewer than expected, "
                      f"please add `**kwargs` to the function signature."
                  )
  
              if iscoroutinefunction(fn):
  
                  @functools.wraps(fn)
                  async def arg_wrapper(
                      executor_instance, *args, **kwargs
                  ):  # we need to get the summary from the executor, so we need to access the self
                      return await fn(
                          executor_instance, *args, **kwargs
                      )
  
                  self.fn = arg_wrapper
              else:
  
                  @functools.wraps(fn)
                  def arg_wrapper(
                      executor_instance, *args, **kwargs
                  ):  # we need to get the summary from the executor, so we need to access the self
                      return fn(executor_instance, *args, **kwargs)
  
                  self.fn = arg_wrapper
  
          def _unwrap_batching_decorator(self, fn):
              if type(fn).__name__ == "DynamicBatchingDecorator":
                  self._batching_decorator = fn
                  return fn.fn
              else:
                  return fn
  
          def _unwrap_write_decorator(self, fn):
              if type(fn).__name__ == "WriteMethodDecorator":
                  self._write_decorator = fn
                  return fn.fn
              else:
                  return fn
  
          def _inject_owner_attrs(
              self, owner, name, request_schema_arg, response_schema_arg
          ):
              if not hasattr(owner, "requests"):
                  owner.requests = {}
  
              from jina.serve.executors import _FunctionWithSchema
  
              fn_with_schema = (
                  _FunctionWithSchema.get_function_with_schema(self.fn)
              )
  
              request_schema_arg = (
                  request_schema_arg
                  if request_schema_arg
                  else fn_with_schema.request_schema
              )
              response_schema_arg = (
                  response_schema_arg
                  if response_schema_arg
                  else fn_with_schema.response_schema
              )
  
              fn_with_schema = _FunctionWithSchema(
                  fn=fn_with_schema.fn,
                  is_generator=fn_with_schema.is_generator,
                  is_singleton_doc=fn_with_schema.is_singleton_doc,
                  is_batch_docs=fn_with_schema.is_batch_docs,
                  parameters_is_pydantic_model=fn_with_schema.parameters_is_pydantic_model,
                  parameters_model=fn_with_schema.parameters_model,
                  request_schema=request_schema_arg,
                  response_schema=response_schema_arg,
              )
              fn_with_schema.validate()
  
              if isinstance(on, (list, tuple)):
                  for o in on:
                      owner.requests_by_class[owner.__name__][
                          o
                      ] = fn_with_schema
              else:
                  owner.requests_by_class[owner.__name__][
                      on or __default_endpoint__
                  ] = fn_with_schema
  
              setattr(owner, name, self.fn)
  
          def __set_name__(self, owner, name):
              _init_requests_by_class(owner)
              if self._batching_decorator:
                  self._batching_decorator._inject_owner_attrs(
                      owner, name
                  )
              if self._write_decorator:
                  self._write_decorator._inject_owner_attrs(owner, name)
              self.fn.class_name = owner.__name__
              self._inject_owner_attrs(
                  owner, name, request_schema, response_schema
              )
  
          def __call__(self, *args, **kwargs):
              # this is needed to make this decorator work in combination with `@requests`
              return self.fn(*args, **kwargs)
  
      if func:
          return FunctionMapper(func)
      else:
          return FunctionMapper
  
file_path: jina/jina/serve/executors/decorators.py
imports: import functools
  import inspect
  import os
  from pathlib import Path
  from typing import Callable, Dict, List, Optional, Sequence, Type, Union
  from jina._docarray import Document, DocumentArray, docarray_v2
  from jina.constants import __cache_path__
  from jina.helper import is_generator, iscoroutinefunction
  from jina.importer import ImportExtensions
label_desc: Use image node_id 5 for calling a global function with example usage: requests(func) and returns: FunctionMapper, FunctionMapper, self, fn, fn, fn, fn, unknown, fn