bolt-python/slack_bolt/kwargs_injection/async_utils.py at main · 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
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
import inspect
import logging
from typing import Callable, Dict, MutableSequence, Optional, Any

from slack_bolt.request.async_request import AsyncBoltRequest
from slack_bolt.response import BoltResponse
from .async_args import AsyncArgs
from slack_bolt.request.payload_utils import (
    to_options,
    to_shortcut,
    to_action,
    to_view,
    to_command,
    to_event,
    to_message,
    to_step,
)
from ..logger.messages import warning_skip_uncommon_arg_name


def build_async_required_kwargs(
    *,
    logger: logging.Logger,
    required_arg_names: MutableSequence[str],
    request: AsyncBoltRequest,
    response: Optional[BoltResponse],
    next_func: Optional[Callable[[], None]] = None,
    this_func: Optional[Callable] = None,
    error: Optional[Exception] = None,  # for error handlers
    next_keys_required: bool = True,  # False for listeners / middleware / error handlers
) -> Dict[str, Any]:
    all_available_args: Dict[str, Any] = {
        "logger": logger,
        "client": request.context.client,
        "req": request,
        "request": request,
        "resp": response,
        "response": response,
        "context": request.context,
        "body": request.body,
        # payload
        "options": to_options(request.body),
        "shortcut": to_shortcut(request.body),
        "action": to_action(request.body),
        "view": to_view(request.body),
        "command": to_command(request.body),
        "event": to_event(request.body),
        "message": to_message(request.body),
        "step": to_step(request.body),
        # utilities
        "ack": request.context.ack,
        "say": request.context.say,
        "respond": request.context.respond,
        "complete": request.context.complete,
        "fail": request.context.fail,
        "set_status": request.context.set_status,
        "set_title": request.context.set_title,
        "set_suggested_prompts": request.context.set_suggested_prompts,
        "get_thread_context": request.context.get_thread_context,
        "save_thread_context": request.context.save_thread_context,
        "say_stream": request.context.say_stream,
        # middleware
        "next": next_func,
        "next_": next_func,  # for the middleware using Python's built-in `next()` function
        # error handler
        "error": error,  # Exception
    }
    if not next_keys_required:
        all_available_args.pop("next")
        all_available_args.pop("next_")

    all_available_args["payload"] = (
        all_available_args["options"]
        or all_available_args["shortcut"]
        or all_available_args["action"]
        or all_available_args["view"]
        or all_available_args["command"]
        or all_available_args["event"]
        or all_available_args["message"]
        or all_available_args["step"]
        or request.body
    )
    for k, v in request.context.items():
        if k not in all_available_args:
            all_available_args[k] = v

    if len(required_arg_names) > 0:
        # To support instance/class methods in a class for listeners/middleware,
        # check if the first argument is either self or cls
        first_arg_name = required_arg_names[0]
        if first_arg_name in {"self", "cls"}:
            required_arg_names.pop(0)
        elif first_arg_name not in all_available_args.keys() and first_arg_name != "args":
            if this_func is None:
                logger.warning(warning_skip_uncommon_arg_name(first_arg_name))
                required_arg_names.pop(0)
            elif inspect.ismethod(this_func):
                # We are sure that we should skip manipulating this arg
                required_arg_names.pop(0)

    kwargs: Dict[str, Any] = {k: v for k, v in all_available_args.items() if k in required_arg_names}
    found_arg_names = kwargs.keys()
    for name in required_arg_names:
        if name == "args":
            if isinstance(request, AsyncBoltRequest):
                kwargs[name] = AsyncArgs(**all_available_args)
            else:
                logger.warning(f"Unknown Request object type detected ({type(request)})")

        elif name not in found_arg_names:
            logger.warning(f"{name} is not a valid argument")
            kwargs[name] = None
    return kwargs