Joseph Doherty
60 lines
1.8 KiB
Python
60 lines
1.8 KiB
Python
"""Client connection options for the async Python wrapper."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
import grpc
|
|
|
|
from .auth import REDACTED, ApiKey
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ClientOptions:
|
|
"""Connection settings for `GatewayClient.connect`."""
|
|
|
|
endpoint: str
|
|
api_key: str | ApiKey | None = None
|
|
plaintext: bool = False
|
|
ca_file: str | None = None
|
|
server_name_override: str | None = None
|
|
|
|
def __post_init__(self) -> None:
|
|
if not self.endpoint:
|
|
raise ValueError("endpoint must not be empty")
|
|
|
|
if self.plaintext and self.ca_file:
|
|
raise ValueError("ca_file cannot be used with plaintext connections")
|
|
|
|
def __repr__(self) -> str:
|
|
api_key = REDACTED if self.api_key else None
|
|
return (
|
|
f"{type(self).__name__}(endpoint={self.endpoint!r}, "
|
|
f"api_key={api_key!r}, plaintext={self.plaintext!r}, "
|
|
f"ca_file={self.ca_file!r}, "
|
|
f"server_name_override={self.server_name_override!r})"
|
|
)
|
|
|
|
|
|
def create_channel(options: ClientOptions) -> grpc.aio.Channel:
|
|
"""Create a plaintext or TLS `grpc.aio` channel from client options."""
|
|
|
|
channel_options: list[tuple[str, str]] = []
|
|
if options.server_name_override:
|
|
channel_options.append(("grpc.ssl_target_name_override", options.server_name_override))
|
|
|
|
if options.plaintext:
|
|
return grpc.aio.insecure_channel(options.endpoint, options=channel_options)
|
|
|
|
root_certificates = None
|
|
if options.ca_file:
|
|
root_certificates = Path(options.ca_file).read_bytes()
|
|
|
|
credentials = grpc.ssl_channel_credentials(root_certificates=root_certificates)
|
|
return grpc.aio.secure_channel(
|
|
options.endpoint,
|
|
credentials,
|
|
options=channel_options,
|
|
)
|