jina.flow.base module

class jina.flow.base.Flow(asyncio: Optional[bool] = False, host: Optional[str] = '0.0.0.0', port_expose: Optional[int] = None, protocol: Optional[str] = 'GRPC', proxy: Optional[bool] = False, **kwargs)[source]
class jina.flow.base.Flow(compress: Optional[str] = 'NONE', compress_min_bytes: Optional[int] = 1024, compress_min_ratio: Optional[float] = 1.1, cors: Optional[bool] = False, ctrl_with_ipc: Optional[bool] = True, daemon: Optional[bool] = False, default_swagger_ui: Optional[bool] = False, description: Optional[str] = None, env: Optional[dict] = None, expose_endpoints: Optional[str] = None, expose_public: Optional[bool] = False, host: Optional[str] = '0.0.0.0', host_in: Optional[str] = '0.0.0.0', host_out: Optional[str] = '0.0.0.0', hosts_in_connect: Optional[List[str]] = None, log_config: Optional[str] = None, memory_hwm: Optional[int] = - 1, name: Optional[str] = 'gateway', no_crud_endpoints: Optional[bool] = False, no_debug_endpoints: Optional[bool] = False, on_error_strategy: Optional[str] = 'IGNORE', port_ctrl: Optional[int] = None, port_expose: Optional[int] = None, port_in: Optional[int] = None, port_out: Optional[int] = None, prefetch: Optional[int] = 50, prefetch_on_recv: Optional[int] = 1, protocol: Optional[str] = 'GRPC', proxy: Optional[bool] = False, py_modules: Optional[List[str]] = None, quiet: Optional[bool] = False, quiet_error: Optional[bool] = False, runtime_backend: Optional[str] = 'PROCESS', runtime_cls: Optional[str] = 'GRPCRuntime', socket_in: Optional[str] = 'PULL_CONNECT', socket_out: Optional[str] = 'PUSH_CONNECT', ssh_keyfile: Optional[str] = None, ssh_password: Optional[str] = None, ssh_server: Optional[str] = None, timeout_ctrl: Optional[int] = 5000, timeout_ready: Optional[int] = 600000, title: Optional[str] = None, uses: Optional[Union[str, Type[BaseExecutor], dict]] = 'BaseExecutor', uses_metas: Optional[dict] = None, uses_with: Optional[dict] = None, workspace: Optional[str] = None, zmq_identity: Optional[str] = None, **kwargs)
class jina.flow.base.Flow(env: Optional[dict] = None, inspect: Optional[str] = 'COLLECT', log_config: Optional[str] = None, name: Optional[str] = None, quiet: Optional[bool] = False, quiet_error: Optional[bool] = False, uses: Optional[str] = None, workspace: Optional[str] = './', **kwargs)

Bases: jina.clients.mixin.PostMixin, jina.jaml.JAMLCompatible, contextlib.ExitStack

Flow is how Jina streamlines and distributes Executors.

property last_pod

Last pod

needs(needs, name='joiner', *args, **kwargs)[source]

Add a blocker to the Flow, wait until all peas defined in needs completed.

Parameters
  • needs (Union[Tuple[str], List[str]]) – list of service names to wait

  • name (str) – the name of this joiner, by default is joiner

  • args – additional positional arguments forwarded to the add function

  • kwargs – additional key value arguments forwarded to the add function

Return type

Flow

Returns

the modified Flow

needs_all(name='joiner', *args, **kwargs)[source]

Collect all hanging Pods so far and add a blocker to the Flow; wait until all handing peas completed.

Parameters
  • name (str) – the name of this joiner (default is joiner)

  • args – additional positional arguments which are forwarded to the add and needs function

  • kwargs – additional key value arguments which are forwarded to the add and needs function

Return type

Flow

Returns

the modified Flow

add(ctrl_with_ipc: Optional[bool] = False, daemon: Optional[bool] = False, docker_kwargs: Optional[dict] = None, entrypoint: Optional[str] = None, env: Optional[dict] = None, expose_public: Optional[bool] = False, external: Optional[bool] = False, gpus: Optional[str] = None, host: Optional[str] = '0.0.0.0', host_in: Optional[str] = '0.0.0.0', host_out: Optional[str] = '0.0.0.0', hosts_in_connect: Optional[List[str]] = None, log_config: Optional[str] = None, memory_hwm: Optional[int] = - 1, name: Optional[str] = None, on_error_strategy: Optional[str] = 'IGNORE', parallel: Optional[int] = 1, peas_hosts: Optional[List[str]] = None, polling: Optional[str] = 'ANY', port_ctrl: Optional[int] = None, port_expose: Optional[int] = None, port_in: Optional[int] = None, port_out: Optional[int] = None, proxy: Optional[bool] = False, pull_latest: Optional[bool] = False, py_modules: Optional[List[str]] = None, quiet: Optional[bool] = False, quiet_error: Optional[bool] = False, quiet_remote_logs: Optional[bool] = False, replicas: Optional[int] = 1, runtime_backend: Optional[str] = 'PROCESS', runtime_cls: Optional[str] = 'ZEDRuntime', scheduling: Optional[str] = 'LOAD_BALANCE', socket_in: Optional[str] = 'PULL_BIND', socket_out: Optional[str] = 'PUSH_BIND', ssh_keyfile: Optional[str] = None, ssh_password: Optional[str] = None, ssh_server: Optional[str] = None, timeout_ctrl: Optional[int] = 5000, timeout_ready: Optional[int] = 600000, upload_files: Optional[List[str]] = None, uses: Optional[Union[str, Type[‘BaseExecutor’], dict]] = 'BaseExecutor', uses_after: Optional[Union[str, Type[‘BaseExecutor’], dict]] = None, uses_before: Optional[Union[str, Type[‘BaseExecutor’], dict]] = None, uses_metas: Optional[dict] = None, uses_with: Optional[dict] = None, volumes: Optional[List[str]] = None, workspace: Optional[str] = None, zmq_identity: Optional[str] = None, **kwargs)Union[‘Flow’, ‘AsyncFlow’][source]

Add a Pod to the current Flow object and return the new modified Flow object. The attribute of the Pod can be later changed with set() or deleted with remove()

Parameters
  • needs (Union[str, Tuple[str], List[str], None]) – the name of the Pod(s) that this Pod receives data from. One can also use ‘gateway’ to indicate the connection with the gateway.

  • pod_role (PodRoleType) – the role of the Pod, used for visualization and route planning

  • copy_flow (bool) – when set to true, then always copy the current Flow and do the modification on top of it then return, otherwise, do in-line modification

  • kwargs – other keyword-value arguments that the Pod CLI supports

Return type

Flow

Returns

a (new) Flow object with modification

inspect(name='inspect', *args, **kwargs)[source]

Add an inspection on the last changed Pod in the Flow

Internally, it adds two Pods to the Flow. But don’t worry, the overhead is minimized and you can remove them by simply using Flow(inspect=FlowInspectType.REMOVE) before using the Flow.

Flow -- PUB-SUB -- BasePod(_pass) -- Flow
        |
        -- PUB-SUB -- InspectPod (Hanging)

In this way, InspectPod looks like a simple _pass from outside and does not introduce side-effects (e.g. changing the socket type) to the original Flow. The original incoming and outgoing socket types are preserved.

This function is very handy for introducing an Evaluator into the Flow.

See also

gather_inspect()

Parameters
  • name (str) – name of the Pod

  • args – args for .add()

  • kwargs – kwargs for .add()

Return type

Flow

Returns

the new instance of the Flow

gather_inspect(name='gather_inspect', include_last_pod=True, *args, **kwargs)[source]

Gather all inspect Pods output into one Pod. When the Flow has no inspect Pod then the Flow itself is returned.

Note

If --no-inspect is not given, then gather_inspect() is auto called before build(). So in general you don’t need to manually call gather_inspect().

Parameters
  • name (str) – the name of the gather Pod

  • include_last_pod (bool) – if to include the last modified Pod in the Flow

  • args – args for .add()

  • kwargs – kwargs for .add()

Return type

Flow

Returns

the modified Flow or the copy of it

See also

inspect()

build(copy_flow=False)[source]

Build the current Flow and make it ready to use

Note

No need to manually call it since 0.0.8. When using Flow with the context manager, or using start(), build() will be invoked.

Parameters

copy_flow (bool) – when set to true, then always copy the current Flow and do the modification on top of it then return, otherwise, do in-line modification

Return type

Flow

Returns

the current Flow (by default)

Note

copy_flow=True is recommended if you are building the same Flow multiple times in a row. e.g.

f = Flow()
with f:
    f.index()

with f.build(copy_flow=True) as fl:
    fl.search()
start()[source]

Start to run all Pods in this Flow.

Remember to close the Flow with close().

Note that this method has a timeout of timeout_ready set in CLI, which is inherited all the way from jina.peapods.peas.BasePea

Returns

this instance

property num_pods

Get the number of Pods in this Flow

Return type

int

property num_peas

Get the number of peas (parallel count) in this Flow

Return type

int

property client

Return a BaseClient object attach to this Flow.

Return type

BaseClient

plot(output=None, vertical_layout=False, inline_display=False, build=True, copy_flow=True)[source]

Visualize the Flow up to the current point If a file name is provided it will create a jpg image with that name, otherwise it will display the URL for mermaid. If called within IPython notebook, it will be rendered inline, otherwise an image will be created.

Example,

flow = Flow().add(name='pod_a').plot('flow.svg')
Parameters
  • output (Optional[str]) – a filename specifying the name of the image to be created, the suffix svg/jpg determines the file type of the output image

  • vertical_layout (bool) – top-down or left-right layout

  • inline_display (bool) – show image directly inside the Jupyter Notebook

  • build (bool) – build the Flow first before plotting, gateway connection can be better showed

  • copy_flow (bool) – when set to true, then always copy the current Flow and do the modification on top of it then return, otherwise, do in-line modification

Return type

Flow

Returns

the Flow

property port_expose

Return the exposed port of the gateway .. # noqa: DAR201

Return type

int

property host

Return the local address of the gateway .. # noqa: DAR201

Return type

str

property address_private

Return the private IP address of the gateway for connecting from other machine in the same network

Return type

str

property address_public

Return the public IP address of the gateway for connecting from other machine in the public network

Return type

str

block()[source]

Block the process until user hits KeyboardInterrupt

property protocol

Return the protocol of this Flow

Return type

GatewayProtocolType

Returns

the protocol of this Flow

property workspace

Return the workspace path of the flow.

Return type

str

property workspace_id

Get all Pods’ workspace_id values in a dict

Return type

Dict[str, str]

property identity

Get all Pods’ identity values in a dict

Return type

Dict[str, str]

expose_endpoint(exec_endpoint: str, path: Optional[str] = None)[source]
expose_endpoint(exec_endpoint: str, *, path: Optional[str] = 'None', status_code: int = '200', tags: Optional[List[str]] = 'None', summary: Optional[str] = 'None', description: Optional[str] = 'None', response_description: str = "'Successful Response'", deprecated: Optional[bool] = 'None', methods: Optional[List[str]] = 'None', operation_id: Optional[str] = 'None', response_model_by_alias: bool = 'True', response_model_exclude_unset: bool = 'False', response_model_exclude_defaults: bool = 'False', response_model_exclude_none: bool = 'False', include_in_schema: bool = 'True', name: Optional[str] = 'None')

Expose an Executor’s endpoint (defined by @requests(on=…)) to HTTP endpoint for easier access.

After expose, you can send data request directly to http://hostname:port/endpoint.

Parameters

exec_endpoint (str) – the endpoint string, by convention starts with /

# noqa: DAR101 # noqa: DAR102

join(needs, name='joiner', *args, **kwargs)

Add a blocker to the Flow, wait until all peas defined in needs completed.

Parameters
  • needs (Union[Tuple[str], List[str]]) – list of service names to wait

  • name (str) – the name of this joiner, by default is joiner

  • args – additional positional arguments forwarded to the add function

  • kwargs – additional key value arguments forwarded to the add function

Return type

Flow

Returns

the modified Flow

rolling_update(pod_name, dump_path=None)[source]

Reload Pods sequentially - only used for compound pods.

Parameters
  • dump_path (Optional[str]) – the path from which to read the dump data

  • pod_name (str) – pod to update

property client_args

Get Client settings.

# noqa: DAR201

Return type

Namespace

property gateway_args

Get Gateway settings.

# noqa: DAR201

Return type

Namespace

update_network_interface(**kwargs)[source]

Update the network interface of this Flow (affects Gateway & Client)

Parameters

kwargs – new network settings