Module slack_bolt.adapter.aws_lambda.lambda_s3_oauth_flow
Expand source code
import logging
import os
from logging import Logger
from typing import Optional
import boto3
from slack_bolt.authorization.authorize import InstallationStoreAuthorize
from slack_bolt.oauth import OAuthFlow
from slack_sdk import WebClient
from slack_sdk.oauth.installation_store.amazon_s3 import AmazonS3InstallationStore
from slack_sdk.oauth.state_store.amazon_s3 import AmazonS3OAuthStateStore
from slack_bolt.oauth.oauth_settings import OAuthSettings
from slack_bolt.util.utils import create_web_client
class LambdaS3OAuthFlow(OAuthFlow):
def __init__(
self,
*,
client: Optional[WebClient] = None,
logger: Optional[Logger] = None,
settings: Optional[OAuthSettings] = None,
oauth_state_bucket_name: Optional[str] = None, # required
installation_bucket_name: Optional[str] = None, # required
):
logger = logger or logging.getLogger(__name__)
settings = settings or OAuthSettings(
client_id=os.environ["SLACK_CLIENT_ID"],
client_secret=os.environ["SLACK_CLIENT_SECRET"],
)
oauth_state_bucket_name = (
oauth_state_bucket_name or os.environ["SLACK_STATE_S3_BUCKET_NAME"]
)
installation_bucket_name = (
installation_bucket_name or os.environ["SLACK_INSTALLATION_S3_BUCKET_NAME"]
)
self.s3_client = boto3.client("s3")
if settings.state_store is None or not isinstance(
settings.state_store, AmazonS3OAuthStateStore
):
settings.state_store = AmazonS3OAuthStateStore(
logger=logger,
s3_client=self.s3_client,
bucket_name=oauth_state_bucket_name,
expiration_seconds=settings.state_expiration_seconds,
)
if settings.installation_store is None or not isinstance(
settings.installation_store, AmazonS3InstallationStore
):
settings.installation_store = AmazonS3InstallationStore(
logger=logger,
s3_client=self.s3_client,
bucket_name=installation_bucket_name,
client_id=settings.client_id,
)
# Set up authorize function to surely use this installation_store.
# When a developer use a settings initialized outside this constructor,
# the settings may already have pre-defined authorize.
# In this case, the /slack/events endpoint doesn't work along with the OAuth flow.
settings.authorize = InstallationStoreAuthorize(
logger=logger,
installation_store=settings.installation_store,
bot_only=settings.installation_store_bot_only,
)
OAuthFlow.__init__(self, client=client, logger=logger, settings=settings)
@property
def client(self) -> WebClient:
if self._client is None:
self._client = create_web_client(logger=self.logger)
return self._client
@property
def logger(self) -> Logger:
if self._logger is None:
self._logger = logging.getLogger(__name__)
return self._logger
Classes
class LambdaS3OAuthFlow (*, client: Optional[slack_sdk.web.client.WebClient] = None, logger: Optional[logging.Logger] = None, settings: Optional[OAuthSettings] = None, oauth_state_bucket_name: Optional[str] = None, installation_bucket_name: Optional[str] = None)
-
The module to run the Slack app installation flow (OAuth flow).
Args
client
- The
slack_sdk.web.WebClient
instance. logger
- The logger.
settings
- OAuth settings to configure this module.
Expand source code
class LambdaS3OAuthFlow(OAuthFlow): def __init__( self, *, client: Optional[WebClient] = None, logger: Optional[Logger] = None, settings: Optional[OAuthSettings] = None, oauth_state_bucket_name: Optional[str] = None, # required installation_bucket_name: Optional[str] = None, # required ): logger = logger or logging.getLogger(__name__) settings = settings or OAuthSettings( client_id=os.environ["SLACK_CLIENT_ID"], client_secret=os.environ["SLACK_CLIENT_SECRET"], ) oauth_state_bucket_name = ( oauth_state_bucket_name or os.environ["SLACK_STATE_S3_BUCKET_NAME"] ) installation_bucket_name = ( installation_bucket_name or os.environ["SLACK_INSTALLATION_S3_BUCKET_NAME"] ) self.s3_client = boto3.client("s3") if settings.state_store is None or not isinstance( settings.state_store, AmazonS3OAuthStateStore ): settings.state_store = AmazonS3OAuthStateStore( logger=logger, s3_client=self.s3_client, bucket_name=oauth_state_bucket_name, expiration_seconds=settings.state_expiration_seconds, ) if settings.installation_store is None or not isinstance( settings.installation_store, AmazonS3InstallationStore ): settings.installation_store = AmazonS3InstallationStore( logger=logger, s3_client=self.s3_client, bucket_name=installation_bucket_name, client_id=settings.client_id, ) # Set up authorize function to surely use this installation_store. # When a developer use a settings initialized outside this constructor, # the settings may already have pre-defined authorize. # In this case, the /slack/events endpoint doesn't work along with the OAuth flow. settings.authorize = InstallationStoreAuthorize( logger=logger, installation_store=settings.installation_store, bot_only=settings.installation_store_bot_only, ) OAuthFlow.__init__(self, client=client, logger=logger, settings=settings) @property def client(self) -> WebClient: if self._client is None: self._client = create_web_client(logger=self.logger) return self._client @property def logger(self) -> Logger: if self._logger is None: self._logger = logging.getLogger(__name__) return self._logger
Ancestors
Class variables
var client_id : str
var failure_handler : Callable[[FailureArgs], BoltResponse]
var install_path : str
var redirect_uri : Optional[str]
var redirect_uri_path : str
var settings : OAuthSettings
var success_handler : Callable[[SuccessArgs], BoltResponse]
Instance variables
var client : slack_sdk.web.client.WebClient
-
Expand source code
@property def client(self) -> WebClient: if self._client is None: self._client = create_web_client(logger=self.logger) return self._client
var logger : logging.Logger
-
Expand source code
@property def logger(self) -> Logger: if self._logger is None: self._logger = logging.getLogger(__name__) return self._logger