Skip to content

Instantly share code, notes, and snippets.

@seratch
Last active January 26, 2022 11:31
Show Gist options
  • Star 2 You must be signed in to star a gist
  • Fork 2 You must be signed in to fork a gist
  • Save seratch/d8933dd4a5dd3fdceeebf30d1564188f to your computer and use it in GitHub Desktop.
Save seratch/d8933dd4a5dd3fdceeebf30d1564188f to your computer and use it in GitHub Desktop.
Two Slack App Installation Flow Example (Flask + SQLAlchemy)
import logging
from typing import Callable
logging.basicConfig(level=logging.DEBUG)
logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)
import os
from slack_bolt import App, BoltContext, Ack
from slack_bolt.adapter.flask import SlackRequestHandler
from slack_bolt.oauth.oauth_settings import OAuthSettings
from slack_sdk import WebClient
from slack_sdk.oauth.installation_store.sqlalchemy import SQLAlchemyInstallationStore
from slack_sdk.oauth.state_store.sqlalchemy import SQLAlchemyOAuthStateStore
import sqlalchemy
from sqlalchemy.engine import Engine
# TODO: change this domain
your_app_domain = "your-own-subdomain.ngrok.io"
logger = logging.getLogger(__name__)
client_id, client_secret, signing_secret = (
os.environ["SLACK_CLIENT_ID"],
os.environ["SLACK_CLIENT_SECRET"],
os.environ["SLACK_SIGNING_SECRET"],
)
#
# Database
#
database_url = "sqlite:///slackapp.db"
engine: Engine = sqlalchemy.create_engine(database_url)
installation_store = SQLAlchemyInstallationStore(
client_id=client_id,
engine=engine,
logger=logger,
)
oauth_state_store = SQLAlchemyOAuthStateStore(
expiration_seconds=120,
engine=engine,
logger=logger,
)
# additional permissions associated with individual users
user_installation_store = SQLAlchemyInstallationStore(
client_id=client_id,
engine=engine,
bots_table_name="user_slack_bots", # just for compatibility with the built-in implementation - we won't use this
installations_table_name="user_slack_installations",
logger=logger,
)
# initialize database tables
try:
engine.execute("select count(*) from slack_bots")
except Exception as e:
installation_store.metadata.create_all(engine)
oauth_state_store.metadata.create_all(engine)
user_installation_store.metadata.create_all(engine)
#
# Main app
#
app = App(
logger=logger,
signing_secret=signing_secret,
oauth_settings=OAuthSettings(
client_id=client_id,
client_secret=client_secret,
installation_store=installation_store,
state_store=oauth_state_store,
scopes=["commands"],
user_scopes=[],
),
)
@app.middleware
def set_user_token_if_exists(context: BoltContext, next: Callable):
installation = user_installation_store.find_installation(
enterprise_id=context.enterprise_id,
team_id=context.team_id,
user_id=context.user_id,
is_enterprise_install=context.is_enterprise_install,
)
if installation is not None:
context["user_token"] = installation.user_token
next()
@app.command("/start-reacting")
def handle_commands(ack: Ack, body: dict, context: BoltContext, client: WebClient):
if context.user_token is None:
ack(
text="Please grant this app to add reactions on behalf of you!",
blocks=[
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "Please grant this app to add reactions on behalf of you!",
},
"accessory": {
"type": "button",
"action_id": "link-button-click",
"text": {"type": "plain_text", "text": "Connect with this app"},
"value": "user_install",
"style": "primary",
"url": f"https://{your_app_domain}/slack/user_install",
},
}
],
)
return
ack()
client.views_open(
trigger_id=body["trigger_id"],
view={
"type": "modal",
"title": {"type": "plain_text", "text": "My App"},
"submit": {"type": "plain_text", "text": "Submit"},
"close": {"type": "plain_text", "text": "Cancel"},
"blocks": [
{
"type": "section",
"text": {"type": "mrkdwn", "text": "Let's start!"},
}
],
},
)
@app.action("link-button-click")
def handle_link_button_clicks(ack: Ack):
ack()
# TODO: delete database rows when receiving tokens_revoked / app_uninstalled events
#
# Web App
#
from flask import Flask, request
flask_app = Flask(__name__)
handler = SlackRequestHandler(app)
@flask_app.route("/slack/events", methods=["POST"])
def slack_events():
return handler.handle(request)
@flask_app.route("/slack/install", methods=["GET"])
def install():
return handler.handle(request)
@flask_app.route("/slack/oauth_redirect", methods=["GET"])
def oauth_redirect():
return handler.handle(request)
user_installation_app = App(
logger=logger,
signing_secret=signing_secret,
oauth_settings=OAuthSettings(
client_id=client_id,
client_secret=client_secret,
installation_store=user_installation_store,
state_store=oauth_state_store,
install_path="/slack/user_install",
redirect_uri_path="/slack/user_oauth_redirect",
redirect_uri=f"https://{your_app_domain}/slack/user_oauth_redirect",
scopes=[],
user_scopes=["reactions:write"],
),
)
user_handler = SlackRequestHandler(user_installation_app)
@flask_app.route("/slack/user_install", methods=["GET"])
def user_install():
return user_handler.handle(request)
@flask_app.route("/slack/user_oauth_redirect", methods=["GET"])
def user_oauth_redirect():
return user_handler.handle(request)
# pip install -r requirements.txt
# # -- OAuth flow -- #
# export SLACK_SIGNING_SECRET=***
# export SLACK_CLIENT_ID=111.111
# export SLACK_CLIENT_SECRET=***
# FLASK_APP=app.py FLASK_ENV=development flask run -p 3000
slack-bolt>=1.4.3,<2
Flask>=1.1,<2
SQLAlchemy
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment