-
Notifications
You must be signed in to change notification settings - Fork 755
/
client.py
80 lines (64 loc) · 2.28 KB
/
client.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
from __future__ import annotations
import asyncio
import functools
from typing import TYPE_CHECKING
import numpy as np
from bentoml.client import Client
if TYPE_CHECKING:
from bentoml.client import GrpcClient
async def arun(c: GrpcClient):
print("registered services:", await c.get_services())
async with c.aservice("health") as health_client:
res = await health_client.Check(service="bentoml.grpc.v1.BentoService")
print(res)
async with c.aservice() as client:
res = await client.Call(
api_name="classify",
ndarray=dict(
dtype=1,
shape=[1, 4],
float_values=[5.9, 3, 5.1, 1.8],
),
)
print("Result from 'await client.Call':\n", res)
res = await client.async_classify(np.array([[5.9, 3, 5.1, 1.8]]))
print("Result from 'client.async_classify':\n", res)
def run(c: Client):
with c.service("health") as health_client:
Check = functools.partial(
health_client.Check, service="bentoml.grpc.v1.BentoService"
)
for to_json in (True, False):
print(
f"Health check ({'serialized' if to_json else 'deserialized'}):",
Check(to_json=to_json),
)
with c.service() as client:
res = client.Call(
api_name="classify",
ndarray=dict(
dtype=1,
shape=[1, 4],
float_values=[5.9, 3, 5.1, 1.8],
),
)
print("Result from 'client.Call':\n", res)
res = client.classify(np.array([[5.9, 3, 5.1, 1.8]]))
print("Result from 'client.classify':\n", res)
res = client.call("classify", np.array([[5.9, 3, 5.1, 1.8]]))
print("Result from 'client.call(bentoml_api_name='classify')':\n", res)
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("-rwa", "--run-with-async", action="store_true")
args = parser.parse_args()
c = Client.from_url("localhost:3000", grpc=True)
if args.run_with_async:
loop = asyncio.new_event_loop()
try:
loop.run_until_complete(arun(c))
finally:
loop.close()
assert loop.is_closed()
else:
run(c)