Module slack_bolt.adapter.socket_mode.internals

Internal functions

Functions

def build_headers(req: slack_sdk.socket_mode.request.SocketModeRequest) ‑> Dict[str, str | Sequence[str]] | None

Expand source code

def build_headers(req: SocketModeRequest) -> Optional[Dict[str, Union[str, Sequence[str]]]]:
    # Mirror the HTTP mode retry headers so middleware/listeners can detect Events API retries
    headers: Dict[str, Union[str, Sequence[str]]] = {}
    if req.retry_attempt is not None:
        headers["x-slack-retry-num"] = str(req.retry_attempt)
    if req.retry_reason is not None:
        headers["x-slack-retry-reason"] = req.retry_reason
    return headers or None

def run_bolt_app(app: App,
req: slack_sdk.socket_mode.request.SocketModeRequest)

Expand source code

def run_bolt_app(app: App, req: SocketModeRequest):
    bolt_req: BoltRequest = BoltRequest(mode="socket_mode", body=req.payload, headers=build_headers(req))
    bolt_resp: BoltResponse = app.dispatch(bolt_req)
    return bolt_resp

def send_response(client: slack_sdk.socket_mode.client.BaseSocketModeClient,
req: slack_sdk.socket_mode.request.SocketModeRequest,
bolt_resp: BoltResponse,
start_time: float)

Expand source code

def send_response(
    client: BaseSocketModeClient,
    req: SocketModeRequest,
    bolt_resp: BoltResponse,
    start_time: float,
):
    if bolt_resp.status == 200:
        content_type = bolt_resp.headers.get("content-type", [""])[0]
        if bolt_resp.body is None or len(bolt_resp.body) == 0:
            client.send_socket_mode_response(SocketModeResponse(envelope_id=req.envelope_id))
        elif content_type.startswith("application/json"):
            dict_body = json.loads(bolt_resp.body)
            client.send_socket_mode_response(SocketModeResponse(envelope_id=req.envelope_id, payload=dict_body))
        else:
            client.send_socket_mode_response(
                SocketModeResponse(envelope_id=req.envelope_id, payload={"text": bolt_resp.body})
            )

        if client.logger.level <= logging.DEBUG:
            spent_time = int((time() - start_time) * 1000)
            client.logger.debug(f"Response time: {spent_time} milliseconds")
    else:
        client.logger.info(f"Unsuccessful Bolt execution result (status: {bolt_resp.status}, body: {bolt_resp.body})")