from .find import FindMixin

from .post import PostMixin

PostMixin,

from typing import TYPE_CHECKING

if TYPE_CHECKING:

from ... import DocumentArray

class PostMixin:

"""Helper functions for posting DocumentArray to Jina Flow. """

def post(self, host: str, show_progress: bool = False) -> 'DocumentArray':

"""Posting itself to a remote Flow/Sandbox and get the modified DocumentArray back

from urllib.parse import urlparse

r = urlparse(host)

_on = r.path or '/'

_port = r.port or None

standardized_host = (

r._replace(netloc=r.netloc.replace(f':{r.port}', ''))

._replace(path='')

.geturl()

)

if r.scheme.startswith('jinahub'):

from jina import Flow

f = Flow(quiet=True).add(uses=standardized_host)

with f:

return f.post(_on, inputs=self, show_progress=show_progress)

elif r.scheme in ('grpc', 'http', 'websocket'):

if _port is None:

raise ValueError(f'can not determine port from {host}')

from jina import Client

c = Client(host=r.hostname, port=_port, protocol=r.scheme)

return c.post(_on, inputs=self, show_progress=show_progress)

else:

raise ValueError(f'unsupported scheme: {r.scheme}')

(da-post)=

You can call an external Flow/Sandbox/Executor to "process" a DocumentArray via {meth}`~docarray.array.mixins.post.PostMixin.post`. The external Flow/Executor can be either local, remote, or inside Docker container.

For example, to use an existing Flow on `192.168.2.3` on port `12345` to process a DocumentArray:

from docarray import DocumentArray

da = DocumentArray.empty(10)

r = da.post('grpc://192.168.2.3:12345')

r.summary()

One can also use any [Executor from Jina Hub](https://hub.jina.ai), e.g.

from docarray import DocumentArray, Document

da = DocumentArray([Document(text='Hi Alex, are you with your laptop?')])

r = da.post('jinahub+sandbox://CoquiTTS7', show_progress=True)

r.summary()

{meth}`~docarray.array.mixins.post.PostMixin.post` accepts a URI-like scheme that supports a wide range of Flow/Hub Executor. It is described as below:

| Attribute | Supported Values | Meaning |

|-----------|---------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------|

| `scheme` | 1. One of `grpc`, `websocket`, `http` | `protocol` of the connected Flow |

| | 2. One of `jinahub`, `jinahub+docker`, `jinhub+sandbox` | Jina hub executor in source code, Docker container, sandbox |

| `netloc` | 1. Host address | `host` of the connected Flow |

| | 2. Hub Executor name | Any Executor [listed here](https://hub.jina.ai) |

| `:port` | e.g. `:55566` | `port` of the connected Flow. This is required when using `scheme` type (1) ; it is ignored when using hub-related `scheme` type (2) |

| `/path` | e.g. `/foo` | The endpoint of the Executor you want to call. |

Some examples:

- `.post('websocket://localhost:8081/foo')`: call the `/foo` endpoint of the Flow on `localhost` port `8081` with `websocket` protocol to process the DocumentArray; processing is on local.

- `.post('grpc://192.168.12.2:12345/foo')`: call the `/foo` endpoint of the Flow on `192.168.12.2` port `12345` with `grpc` protocol to process the DocumentArray; processing is on remote.

- `.post('jinahub://Hello/foo')`: call the `/foo` endpoint of the Hub Executor `Hello` to process the DocumentArray; porcessing is on local.

- `.post('jinahub+sandbox://Hello/foo')`: call the `/foo` endpoint of the Hub Sandbox `Hello` to process the DocumentArray; porcessing is on remote.

For more explanation of Flow, Hub Executor and Sandbox, please refer to [Jina docs](https://docs.jina.ai).

As described {ref}`here <da-post>`, one can simply use an external Jina Flow/Executor as a regular function to process a DocumentArray.

'jina',

import multiprocessing

import time

import pytest

from docarray import DocumentArray

from docarray.helper import random_port

def test_post_to_a_flow(show_pbar, conn_config):

from jina import Flow

def start_flow(stop_event, **kwargs):

"""start a blocking Flow."""

with Flow(**kwargs) as f:

f.block(stop_event=stop_event)

print('bye')

e = multiprocessing.Event() # create new Event

p = random_port()

t = multiprocessing.Process(

name='Blocked-Flow',

target=start_flow,

args=(e,),

kwargs={**conn_config[0], 'port': p},

)

t.start()

time.sleep(1)

da = DocumentArray.empty(100)

try:

da.post(conn_config[1].replace('$port', str(p)))

except:

raise

finally:

e.set()

t.join()

time.sleep(1)

def test_post_with_jinahub(hub_uri):

da = DocumentArray.empty(100)

da.post(hub_uri)

def test_post_bad_scheme():

da = DocumentArray.empty(100)

with pytest.raises(ValueError):

da.post('haha')

def test_endpoint():

from jina import Executor, requests, Flow

class MyExec(Executor):

@requests(on='/foo')

def foo(self, docs: DocumentArray, **kwargs):

docs.texts = ['foo'] * len(docs)

@requests(on='/bar')

def bar(self, docs: DocumentArray, **kwargs):

docs.texts = ['bar'] * len(docs)

def start_flow(stop_event, **kwargs):

"""start a blocking Flow."""

with Flow(**kwargs).add(uses=MyExec) as f:

f.block(stop_event=stop_event)

e = multiprocessing.Event() # create new Event

p = random_port()

t = multiprocessing.Process(

name='Blocked-Flow', target=start_flow, args=(e,), kwargs={'port': p}

)

t.start()

time.sleep(1)

N = 100

da = DocumentArray.empty(N)

try:

assert da.post(f'grpc://0.0.0.0:{p}/')[:, 'text'] == [''] * N

assert da.post(f'grpc://0.0.0.0:{p}/foo').texts == ['foo'] * N

assert da.post(f'grpc://0.0.0.0:{p}/bar').texts == ['bar'] * N

except:

raise

finally:

e.set()

t.join()