bedrock.websockets.websocket_gateway_factory

 1from bedrock.config import get_config_params
 2from bedrock.config._common import validate_config
 3from bedrock.endpoints.local_websocket_api_gateway import LocalWebsocketApiGateway
 4from bedrock.external.api_gateway import init_aws_websocket_api_gateway_client
 5
 6GATEWAY = None
 7
 8
 9def get_websocket_api_gateway_client():  # pragma: integration
10    """
11    Returns the correct WebSocket API Gateway client based on environment.
12    Caches the client after the first creation.
13    """
14    global GATEWAY
15    if GATEWAY is None:
16        GATEWAY = _init_websocket_api_gateway_client()
17    return GATEWAY
18
19
20def _init_websocket_api_gateway_client():  # pragma: integration
21    config = get_config_params()
22    _validate_websocket_api_gateway_config(config)
23    gateway_id = config["aws"]["api_gateway"]["gateway_id"]
24    if gateway_id == "local":
25        return LocalWebsocketApiGateway()
26
27    return init_aws_websocket_api_gateway_client()
28
29
30def _validate_websocket_api_gateway_config(config: dict):  # pragma: integration
31    validate_config(config["aws"].get("api_gateway", {}), ["gateway_id"], "aws.api_gateway.")