Module slack_bolt.middleware.authorization.async_single_team_authorization

Classes

class AsyncSingleTeamAuthorization (base_logger: Optional[logging.Logger] = None, user_facing_authorize_error_message: Optional[str] = None)

A middleware can process request data before other middleware and listener functions.

Single-workspace authorization.

Expand source code 
class AsyncSingleTeamAuthorization(AsyncAuthorization):
    def __init__(
        self,
        base_logger: Optional[Logger] = None,
        user_facing_authorize_error_message: Optional[str] = None,
    ):
        """Single-workspace authorization."""
        self.auth_test_result: Optional[AsyncSlackResponse] = None
        self.logger = get_bolt_logger(AsyncSingleTeamAuthorization, base_logger=base_logger)
        self.user_facing_authorize_error_message = (
            user_facing_authorize_error_message or _build_user_facing_authorize_error_message()
        )

    async def async_process(
        self,
        *,
        req: AsyncBoltRequest,
        resp: BoltResponse,
        # As this method is not supposed to be invoked by bolt-python users,
        # the naming conflict with the built-in one affects
        # only the internals of this method
        next: Callable[[], Awaitable[BoltResponse]],
    ) -> BoltResponse:
        if _is_no_auth_required(req):
            return await next()

        if _is_no_auth_test_call_required(req):
            req.context.set_authorize_result(
                AuthorizeResult(
                    enterprise_id=req.context.enterprise_id,
                    team_id=req.context.team_id,
                    user_id=req.context.user_id,
                )
            )
            return await next()

        try:
            if self.auth_test_result is None:
                self.auth_test_result = await req.context.client.auth_test()

            if self.auth_test_result:
                req.context.set_authorize_result(
                    _to_authorize_result(
                        auth_test_result=self.auth_test_result,
                        token=req.context.client.token,
                        request_user_id=req.context.user_id,
                    )
                )
                return await next()
            else:
                # Just in case
                self.logger.error("auth.test API call result is unexpectedly None")
                if req.context.response_url is not None:
                    await req.context.respond(self.user_facing_authorize_error_message)  # type: ignore[misc]
                    return BoltResponse(status=200, body="")
                return _build_user_facing_error_response(self.user_facing_authorize_error_message)
        except SlackApiError as e:
            self.logger.error(f"Failed to authorize with the given token ({e})")
            return _build_user_facing_error_response(self.user_facing_authorize_error_message)

Ancestors

Inherited members