-
Notifications
You must be signed in to change notification settings - Fork 234
Expand file tree
/
Copy pathgrpcio_server.py
More file actions
34 lines (26 loc) · 759 Bytes
/
Copy pathgrpcio_server.py
File metadata and controls
34 lines (26 loc) · 759 Bytes
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
from typing import Dict, TYPE_CHECKING
from abc import ABC, abstractmethod
if TYPE_CHECKING:
import grpc
class ServicerBase(ABC):
"""
Base class for async grpcio servers.
"""
@property
@abstractmethod
def __rpc_methods__(self) -> Dict[str, "grpc.RpcMethodHandler"]:
...
@property
@abstractmethod
def __proto_path__(self) -> str:
...
def register_servicers(server: "grpc.aio.Server", *servicers: ServicerBase):
from grpc import method_handlers_generic_handler
server.add_generic_rpc_handlers(
tuple(
method_handlers_generic_handler(
servicer.__proto_path__, servicer.__rpc_handlers__
)
for servicer in servicers
)
)