bolt-python/slack_bolt/request/async_internals.py at 1e4939772b9e9cb7d40f1ebf9d3f501685c70cb2 · slackapi/bolt-python · GitHub

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
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
from typing import Dict, Any, Optional, Sequence

from slack_bolt.context.async_context import AsyncBoltContext
from slack_bolt.request.internals import (
    extract_enterprise_id,
    extract_function_bot_access_token,
    extract_function_execution_id,
    extract_function_inputs,
    extract_is_enterprise_install,
    extract_team_id,
    extract_user_id,
    extract_channel_id,
    debug_multiple_response_urls_detected,
    extract_actor_enterprise_id,
    extract_actor_team_id,
    extract_actor_user_id,
    extract_thread_ts,
)


def build_async_context(
    context: AsyncBoltContext,
    body: Dict[str, Any],
    headers: Optional[Dict[str, Sequence[str]]] = None,
) -> AsyncBoltContext:
    context["is_enterprise_install"] = extract_is_enterprise_install(body)
    enterprise_id = extract_enterprise_id(body)
    if enterprise_id:
        context["enterprise_id"] = enterprise_id
    team_id = extract_team_id(body)
    if team_id:
        context["team_id"] = team_id
    user_id = extract_user_id(body)
    if user_id:
        context["user_id"] = user_id
    # Actor IDs are useful for Events API on a Slack Connect channel
    actor_enterprise_id = extract_actor_enterprise_id(body)
    if actor_enterprise_id:
        context["actor_enterprise_id"] = actor_enterprise_id
    actor_team_id = extract_actor_team_id(body)
    if actor_team_id:
        context["actor_team_id"] = actor_team_id
    actor_user_id = extract_actor_user_id(body)
    if actor_user_id:
        context["actor_user_id"] = actor_user_id
    channel_id = extract_channel_id(body)
    if channel_id:
        context["channel_id"] = channel_id
    thread_ts = extract_thread_ts(body)
    if thread_ts:
        context["thread_ts"] = thread_ts
    function_execution_id = extract_function_execution_id(body)
    if function_execution_id:
        context["function_execution_id"] = function_execution_id
        function_bot_access_token = extract_function_bot_access_token(body)
        if function_bot_access_token is not None:
            context["function_bot_access_token"] = function_bot_access_token
        function_inputs = extract_function_inputs(body)
        if function_inputs is not None:
            context["inputs"] = function_inputs
    if "response_url" in body:
        context["response_url"] = body["response_url"]
    elif "response_urls" in body:
        # In the case where response_url_enabled: true in a modal exists
        response_urls = body["response_urls"]
        if len(response_urls) >= 1:
            if len(response_urls) > 1:
                context.logger.debug(debug_multiple_response_urls_detected())
            response_url = response_urls[0].get("response_url")
            context["response_url"] = response_url

    if headers is not None:
        retry_num_header = headers.get("x-slack-retry-num")
        if retry_num_header is not None and len(retry_num_header) > 0:
            try:
                context["retry_num"] = int(retry_num_header[0])
            except (ValueError, TypeError):
                pass

        retry_reason_header = headers.get("x-slack-retry-reason")
        if retry_reason_header is not None and len(retry_reason_header) > 0:
            context["retry_reason"] = retry_reason_header[0]

    return context