code: def test_call_from_requests_as_singleton(
      ctxt_manager, include_gateway
  ):
      if not include_gateway and ctxt_manager == "flow":
          return
  
      class MySingletonReturnInputDoc(BaseDoc):
          text: str
          price: int
  
      class MySingletonReturnOutputDoc(BaseDoc):
          text: str
          category: str
  
      class MySingletonExecutorReturn(Executor):
          @requests(on="/foo")
          def foo(
              self, docs: DocList[MySingletonReturnInputDoc], **kwargs
          ) -> DocList[MySingletonReturnOutputDoc]:
              ret = DocList[MySingletonReturnOutputDoc]()
              for doc in docs:
                  ret.append(
                      MySingletonReturnOutputDoc(
                          text=doc.text + "_changed",
                          category=str(doc.price + 1),
                      )
                  )
              return ret
  
          @requests(on="/foo_single")
          def foo_single(
              self, doc: MySingletonReturnInputDoc, **kwargs
          ) -> MySingletonReturnOutputDoc:
              return MySingletonReturnOutputDoc(
                  text=doc.text + "_changed",
                  category=str(doc.price + 1),
              )
  
      port = random_port()
  
      if ctxt_manager == "flow":
          ctxt = Flow(port=port, protocol="http").add(
              uses=MySingletonExecutorReturn
          )
      else:
          ctxt = Deployment(
              port=port,
              protocol="http",
              uses=MySingletonExecutorReturn,
              include_gateway=include_gateway,
          )
  
      with ctxt:
          import requests as global_requests
  
          for endpoint in {"foo", "foo_single"}:
              url = f"http://localhost:{port}/{endpoint}"
              myobj = {"data": {"text": "hello", "price": 2}}
              resp = global_requests.post(url, json=myobj)
              resp_json = resp.json()
              assert resp_json["data"][0]["text"] == "hello_changed"
              assert resp_json["data"][0]["category"] == str(3)
              myobj = {"data": [{"text": "hello", "price": 2}]}
              resp = global_requests.post(url, json=myobj)
              resp_json = resp.json()
              assert resp_json["data"][0]["text"] == "hello_changed"
              assert resp_json["data"][0]["category"] == str(3)
              myobj = {
                  "data": [
                      {"text": "hello", "price": 2},
                      {"text": "hello", "price": 2},
                  ]
              }
              resp = global_requests.post(url, json=myobj)
              resp_json = resp.json()
              assert len(resp_json["data"]) == 2
              for d in resp_json["data"]:
                  assert d["text"] == "hello_changed"
                  assert d["category"] == str(3)
  
file_path: jina/tests/integration/docarray_v2/test_singleton.py
imports: import pytest
  from jina import Flow, Executor, Deployment, requests, Client
  from docarray import BaseDoc, DocList
  from docarray.documents import TextDoc, ImageDoc
  from jina.helper import random_port
  from jina.excepts import RuntimeFailToStart
  import asyncio
label_desc: Use image node_id 6 for calling a global function with example usage: test_call_from_requests_as_singleton(ctxt_manager, include_gateway) and returns: ret, MySingletonReturnOutputDoc