Repository: jina-ai/serve
Pull request: feat: add client to deployment
based on: https://github.com/jina-ai/jina/pull/5574
closes: https://github.com/jina-ai/jina/issues/5551

Problem statement:
Offer a client as part of Deployment when serving an Executor
When using a Deployment with the context manager, offer a client as well.
Basically, Deployment should extend PostMixin and by default, expose a gateway (like in Flow).
(This might mean adding more gateway args to deployment.)
In Flow, all usages of Deployment should not add a gateway

Hints from issue discussion:

Commit message: style: fix overload and cli autocomplete

Patch to review:
diff --git a/docs/concepts/executor/serve.md b/docs/concepts/executor/serve.md
index ca50c4a69ef68..b572c4cecb15c 100644
--- a/docs/concepts/executor/serve.md
+++ b/docs/concepts/executor/serve.md
@@ -44,7 +44,6 @@ In other words:
 This separation also aims to enhance the reusability of Executors: the same implementation of an Executor can be 
 served in multiple ways/configurations using Deployment.
 
-Serve the Executor:
 ````{tab} Python class
 
 ```python
@@ -59,7 +58,8 @@ class MyExec(Executor):
 
 
 with Deployment(uses=MyExec, port=12345, replicas=2) as dep:
-    dep.block()
+    docs = dep.post(inputs=DocumentArray.empty(1), on='/foo')
+    print(docs.texts)
 ```
 ````
 
@@ -75,7 +75,8 @@ py_modules:
 from jina import Deployment
 
 with Deployment(uses='executor.yaml', port=12345, replicas=2) as dep:
-    dep.block()
+    docs = dep.post(inputs=DocumentArray.empty(1), on='/foo')
+    print(docs.texts)
 ```
 ````
 
@@ -85,7 +86,8 @@ with Deployment(uses='executor.yaml', port=12345, replicas=2) as dep:
 from jina import Deployment
 
 with Deployment(uses='jinaai://my-username/MyExec/', port=12345, replicas=2) as dep:
-    dep.block()
+    docs = dep.post(inputs=DocumentArray.empty(1), on='/foo')
+    print(docs.texts)
 ```
 
 ````
@@ -96,7 +98,8 @@ with Deployment(uses='jinaai://my-username/MyExec/', port=12345, replicas=2) as
 from jina import Deployment
 
 with Deployment(uses='docker://my-executor-image', port=12345, replicas=2) as dep:
-    dep.block()
+    docs = dep.post(inputs=DocumentArray.empty(1), on='/foo')
+    print(docs.texts)
 ```
 
 ````
@@ -109,20 +112,17 @@ with Deployment(uses='docker://my-executor-image', port=12345, replicas=2) as de
 │  🔒     Private     192.168.3.147:12345   │
 │  🌍      Public    87.191.159.105:12345   │
 ╰───────────────────────────────────────────╯
+['executed MyExec']
 ```
 
-Access the served Executor:
+````{hint}
+You can use `dep.block()` to serve forever:
 
 ```python
-from jina import Client, DocumentArray, Document
-
-print(Client(port=12345).post(inputs=DocumentArray.empty(1), on='/foo').texts)
-```
-
-```shell
-['executed MyExec']
+with Deployment(uses=MyExec, port=12345, replicas=2) as dep:
+    dep.block()
 ```
-
+````
 
 The {class}`~jina.orchestrate.deployments.Deployment` class accepts configuration options similar to 
 {ref}`Executor configuration with Flows <flow-configure-executors>`.
diff --git a/jina/orchestrate/deployments/__init__.py b/jina/orchestrate/deployments/__init__.py
index 8c86f106cfd42..864b2542a98bc 100644
--- a/jina/orchestrate/deployments/__init__.py
+++ b/jina/orchestrate/deployments/__init__.py
@@ -16,6 +16,8 @@
 from rich import print
 from rich.panel import Panel
 
+from jina.clients import Client
+from jina.clients.mixin import PostMixin
 from jina.constants import (
     __default_executor__,
     __default_host__,
@@ -41,10 +43,11 @@
 if TYPE_CHECKING:
     import multiprocessing
 
+    from jina.clients.base import BaseClient
     from jina.serve.executors import BaseExecutor
 
 
-class Deployment(BaseOrchestrator):
+class Deployment(PostMixin, BaseOrchestrator):
     """A Deployment is an immutable set of pods, which run in replicas. They share the same input and output socket.
     Internally, the pods can run with the process/thread backend. They can be also run in their own containers
     :param args: arguments parsed from the CLI
@@ -287,18 +290,18 @@ def __init__(
         **kwargs,
     ):
         super().__init__()
-        gateway_kwargs = {}
+        self._gateway_kwargs = {}
         self._include_gateway = include_gateway
         if self._include_gateway:
             # arguments exclusive to the gateway
             for field in ['port']:
                 if field in kwargs:
-                    gateway_kwargs[field] = kwargs.pop(field)
+                    self._gateway_kwargs[field] = kwargs.pop(field)
 
             # arguments common to both gateway and the Executor
             for field in ['host']:
                 if field in kwargs:
-                    gateway_kwargs[field] = kwargs[field]
+                    self._gateway_kwargs[field] = kwargs[field]
 
         parser = set_deployment_parser()
         if args is None:
@@ -344,7 +347,9 @@ def __init__(
 
         if self._include_gateway:
             gateway_parser = set_gateway_parser()
-            args = ArgNamespace.kwargs2namespace(gateway_kwargs, gateway_parser, True)
+            args = ArgNamespace.kwargs2namespace(
+                self._gateway_kwargs, gateway_parser, True
+            )
 
             args.deployments_addresses = f'{{"executor": ["0.0.0.0:{self.port}"]}}'
             args.graph_description = (
@@ -499,6 +504,20 @@ def head_port_monitoring(self):
         """
         return self.head_args.port_monitoring if self.head_args else None
 
+    @property
+    def client(self) -> 'BaseClient':
+        """Return a :class:`BaseClient` object attach to this Flow.
+
+        .. # noqa: DAR201"""
+
+        kwargs = dict(
+            host=self.host,
+            port=self.port,
+            protocol=self.protocol,
+        )
+        kwargs.update(self._gateway_kwargs)
+        return Client(**kwargs)
+
     @staticmethod
     def _copy_to_head_args(args: Namespace) -> Namespace:
         """
diff --git a/tests/integration/deployments/test_deployment.py b/tests/integration/deployments/test_deployment.py
index 885ac5d5652a2..182f5a0946233 100644
--- a/tests/integration/deployments/test_deployment.py
+++ b/tests/integration/deployments/test_deployment.py
@@ -438,5 +438,19 @@ def serve_depl(stop_event, **kwargs):
 @pytest.mark.parametrize('served_depl', [False, True], indirect=True)
 def test_deployment_dynamic_batching(served_depl, exposed_port):
     docs = Client(port=exposed_port).post(on='/bar', inputs=DocumentArray.empty(5))
+    assert docs.texts == ['bar' for _ in docs]
+
+
+@pytest.mark.parametrize('enable_dynamic_batching', [False, True])
+def test_deployment_client_dynamic_batching(enable_dynamic_batching):
+    kwargs = {'port': random_port()}
+    if enable_dynamic_batching:
+        kwargs['uses_dynamic_batching'] = {
+            '/bar': {'preferred_batch_size': 4, 'timeout': 5000}
+        }
+
+    depl = Deployment(uses=MyServeExec, **kwargs)
+    with depl:
+        docs = depl.post(on='/bar', inputs=DocumentArray.empty(5))
 
     assert docs.texts == ['bar' for _ in docs]
diff --git a/tests/integration/reload/test_deployments_reload.py b/tests/integration/reload/test_deployments_reload.py
index 2a8242ee0d8b5..788df9a797ead 100644
--- a/tests/integration/reload/test_deployments_reload.py
+++ b/tests/integration/reload/test_deployments_reload.py
@@ -4,10 +4,7 @@
 import threading
 import time
 
-import pytest
-
-from jina import Client, DocumentArray, Executor, Flow, requests, Deployment
-from jina.helper import random_port
+from jina import Client, Deployment, DocumentArray, Flow
 
 cur_dir = os.path.dirname(__file__)
 
@@ -80,8 +77,7 @@ def test_deployment_reload(tmpdir):
     t.start()
     time.sleep(5)
     try:
-        client = Client(port=depl.port, protocol=str(depl.protocol))
-        res = client.post(on='/', inputs=DocumentArray.empty(10))
+        res = depl.post(on='/', inputs=DocumentArray.empty(10))
         assert len(res) == 10
         for doc in res:
             assert doc.text == 'MyExecutorBeforeReload'
@@ -90,13 +86,11 @@ def test_deployment_reload(tmpdir):
             os.path.join(os.path.join(cur_dir, 'exec'), 'config.yml'),
             str(tmpdir),
         ):
-            client = Client(port=depl.port, protocol=str(depl.protocol))
-            res = client.post(on='/', inputs=DocumentArray.empty(10))
+            res = depl.post(on='/', inputs=DocumentArray.empty(10))
             assert len(res) == 10
             for doc in res:
                 assert doc.text == 'MyExecutorAfterReload'
-        client = Client(port=depl.port, protocol=str(depl.protocol))
-        res = client.post(on='/', inputs=DocumentArray.empty(10))
+        res = depl.post(on='/', inputs=DocumentArray.empty(10))
         assert len(res) == 10
         for doc in res:
             assert doc.text == 'MyExecutorBeforeReload'


Code review comments:
diff_hunk: @@ -75,7 +75,8 @@ py_modules:
 from jina import Deployment
 
 with Deployment(uses='executor.yaml', port=12345, replicas=2) as dep:
-    dep.block()
+    docs = dep.post(inputs=DocumentArray.empty(1), on='/foo')
path: docs/concepts/executor/serve.md
text: @author:
```suggestion
    docs = dep.post(on='/foo', inputs=DocumentArray.empty(1))
```

diff_hunk: @@ -85,7 +86,8 @@ with Deployment(uses='executor.yaml', port=12345, replicas=2) as dep:
 from jina import Deployment
 
 with Deployment(uses='jinaai://my-username/MyExec/', port=12345, replicas=2) as dep:
-    dep.block()
+    docs = dep.post(inputs=DocumentArray.empty(1), on='/foo')
path: docs/concepts/executor/serve.md
text: @author:
```suggestion
    docs = dep.post(on='/foo', inputs=DocumentArray.empty(1))
```

diff_hunk: @@ -59,7 +58,8 @@ class MyExec(Executor):
 
 
 with Deployment(uses=MyExec, port=12345, replicas=2) as dep:
-    dep.block()
+    docs = dep.post(inputs=DocumentArray.empty(1), on='/foo')
path: docs/concepts/executor/serve.md
text: @author:
```suggestion
    docs = dep.post(on='/foo', inputs=DocumentArray.empty(1))
```

diff_hunk: @@ -96,7 +98,8 @@ with Deployment(uses='jinaai://my-username/MyExec/', port=12345, replicas=2) as
 from jina import Deployment
 
 with Deployment(uses='docker://my-executor-image', port=12345, replicas=2) as dep:
-    dep.block()
+    docs = dep.post(inputs=DocumentArray.empty(1), on='/foo')
path: docs/concepts/executor/serve.md
text: @author:
```suggestion
    docs = dep.post(on='/foo', inputs=DocumentArray.empty(1))
```

diff_hunk: @@ -59,7 +58,8 @@ class MyExec(Executor):
 
 
 with Deployment(uses=MyExec, port=12345, replicas=2) as dep:
-    dep.block()
+    docs = dep.post(inputs=DocumentArray.empty(1), on='/foo')
path: docs/concepts/executor/serve.md
text: @user1:
put on in front, which is more aligned with the typical use of Flow