-
Notifications
You must be signed in to change notification settings - Fork 5
Description
What's needed?
We need to standarize how we create clients, how to pass the destination to connect too, and how we close/cleanup the clients.
Proposed solution
Create a new base class called GrpcApiClient with at least some basic functionality, like:
class GrpcApiClient(Generic[T]):
def __init__(self, target: str, create_stub: Callable[[GrpcChannel], T]):
self._target = target
self._channel = grpcaio.insecure_channel(target)
self._stub: T = create_stub(self._grpc_channel)
async def close(self, timeout: timedelta | None = None) -> None:
await self._channel.close(grace=timeout.total_seconds() if timeout else None)
async def __aenter__(self) -> "ReportingClient":
return self
async def __aexit__(
self,
_exc_type: Type[BaseException] | None,
_exc_val: BaseException | None,
_exc_tb: Any | None,
) -> bool | None:
await self.close()
return NoneUse cases
No response
Alternatives and workarounds
Some clients take a grpc channel, and some a connection string. The advantage of taking a channel is users can chose which type of channel to use (insecure vs. secure for example), the disadvantage is we'll leak the dependency to grpc.
We could extend the connection string to chose if the channel should be secure or insecure, so we could have the best of both world. This would also enable to take connection strings directly from configuration files.
We might need a bit more research, maybe something like that already exists.