Module slack_sdk.oauth.installation_store.sqlite3

Classes

class SQLite3InstallationStore (*, database: str, client_id: str, logger: logging.Logger = <Logger slack_sdk.oauth.installation_store.sqlite3 (WARNING)>)

The installation store interface.

The minimum required methods are:

  • save(installation)
  • find_installation(enterprise_id, team_id, user_id, is_enterprise_install)

If you would like to properly handle app uninstallations and token revocations, the following methods should be implemented.

  • delete_installation(enterprise_id, team_id, user_id)
  • delete_all(enterprise_id, team_id)

If your app needs only bot scope installations, the simpler way to implement would be:

  • save(installation)
  • find_bot(enterprise_id, team_id, is_enterprise_install)
  • delete_bot(enterprise_id, team_id)
  • delete_all(enterprise_id, team_id)
Expand source code
class SQLite3InstallationStore(InstallationStore, AsyncInstallationStore):
    def __init__(
        self,
        *,
        database: str,
        client_id: str,
        logger: Logger = logging.getLogger(__name__),
    ):
        self.database = database
        self.client_id = client_id
        self.init_called = False
        self._logger = logger

    @property
    def logger(self) -> Logger:
        if self._logger is None:
            self._logger = logging.getLogger(__name__)
        return self._logger

    def init(self):
        try:
            with sqlite3.connect(database=self.database) as conn:
                cur = conn.execute("select count(1) from slack_installations;")
                row_num = cur.fetchone()[0]
                self.logger.debug(f"{row_num} installations are stored in {self.database}")
        except Exception:  # skipcq: PYL-W0703
            self.create_tables()
        self.init_called = True

    def connect(self) -> Connection:
        if not self.init_called:
            self.init()
        return sqlite3.connect(database=self.database)

    def create_tables(self):
        with sqlite3.connect(database=self.database) as conn:
            conn.execute(
                """
            create table slack_installations (
                id integer primary key autoincrement,
                client_id text not null,
                app_id text not null,
                enterprise_id text not null default '',
                enterprise_name text,
                enterprise_url text,
                team_id text not null default '',
                team_name text,
                bot_token text,
                bot_id text,
                bot_user_id text,
                bot_scopes text,
                bot_refresh_token text,  -- since v3.8
                bot_token_expires_at datetime,  -- since v3.8
                user_id text not null,
                user_token text,
                user_scopes text,
                user_refresh_token text,  -- since v3.8
                user_token_expires_at datetime,  -- since v3.8
                incoming_webhook_url text,
                incoming_webhook_channel text,
                incoming_webhook_channel_id text,
                incoming_webhook_configuration_url text,
                is_enterprise_install boolean not null default 0,
                token_type text,
                installed_at datetime not null default current_timestamp
            );
            """
            )
            conn.execute(
                """
            create index slack_installations_idx on slack_installations (
                client_id,
                enterprise_id,
                team_id,
                user_id,
                installed_at
            );
            """
            )
            conn.execute(
                """
            create table slack_bots (
                id integer primary key autoincrement,
                client_id text not null,
                app_id text not null,
                enterprise_id text not null default '',
                enterprise_name text,
                team_id text not null default '',
                team_name text,
                bot_token text not null,
                bot_id text not null,
                bot_user_id text not null,
                bot_scopes text,
                bot_refresh_token text,  -- since v3.8
                bot_token_expires_at datetime,  -- since v3.8
                is_enterprise_install boolean not null default 0,
                installed_at datetime not null default current_timestamp
            );
            """
            )
            conn.execute(
                """
            create index slack_bots_idx on slack_bots (
                client_id,
                enterprise_id,
                team_id,
                installed_at
            );
            """
            )
            self.logger.debug(f"Tables have been created (database: {self.database})")
            conn.commit()

    async def async_save(self, installation: Installation):
        return self.save(installation)

    async def async_save_bot(self, bot: Bot):
        return self.save_bot(bot)

    def save(self, installation: Installation):
        with self.connect() as conn:
            conn.execute(
                """
                insert into slack_installations (
                    client_id,
                    app_id,
                    enterprise_id,
                    enterprise_name,
                    enterprise_url,
                    team_id,
                    team_name,
                    bot_token,
                    bot_id,
                    bot_user_id,
                    bot_scopes,
                    bot_refresh_token,  -- since v3.8
                    bot_token_expires_at,  -- since v3.8
                    user_id,
                    user_token,
                    user_scopes,
                    user_refresh_token,  -- since v3.8
                    user_token_expires_at,  -- since v3.8
                    incoming_webhook_url,
                    incoming_webhook_channel,
                    incoming_webhook_channel_id,
                    incoming_webhook_configuration_url,
                    is_enterprise_install,
                    token_type
                )
                values
                (
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?
                );
                """,
                [
                    self.client_id,
                    installation.app_id,
                    installation.enterprise_id or "",
                    installation.enterprise_name,
                    installation.enterprise_url,
                    installation.team_id or "",
                    installation.team_name,
                    installation.bot_token,
                    installation.bot_id,
                    installation.bot_user_id,
                    ",".join(installation.bot_scopes),
                    installation.bot_refresh_token,
                    installation.bot_token_expires_at,
                    installation.user_id,
                    installation.user_token,
                    ",".join(installation.user_scopes) if installation.user_scopes else None,
                    installation.user_refresh_token,
                    installation.user_token_expires_at,
                    installation.incoming_webhook_url,
                    installation.incoming_webhook_channel,
                    installation.incoming_webhook_channel_id,
                    installation.incoming_webhook_configuration_url,
                    1 if installation.is_enterprise_install else 0,
                    installation.token_type,
                ],
            )
            self.logger.debug(
                f"New rows in slack_bots and slack_installations have been created (database: {self.database})"
            )
            conn.commit()

        self.save_bot(installation.to_bot())

    def save_bot(self, bot: Bot):
        if bot.bot_token is None:
            self.logger.debug("Skipped saving a new row because of the absense of bot token in it")
            return

        with self.connect() as conn:
            conn.execute(
                """
                insert into slack_bots (
                    client_id,
                    app_id,
                    enterprise_id,
                    enterprise_name,
                    team_id,
                    team_name,
                    bot_token,
                    bot_id,
                    bot_user_id,
                    bot_scopes,
                    bot_refresh_token,  -- since v3.8
                    bot_token_expires_at,  -- since v3.8
                    is_enterprise_install
                )
                values
                (
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?,
                    ?
                );
                """,
                [
                    self.client_id,
                    bot.app_id,
                    bot.enterprise_id or "",
                    bot.enterprise_name,
                    bot.team_id or "",
                    bot.team_name,
                    bot.bot_token,
                    bot.bot_id,
                    bot.bot_user_id,
                    ",".join(bot.bot_scopes),
                    bot.bot_refresh_token,
                    bot.bot_token_expires_at,
                    bot.is_enterprise_install,
                ],
            )
            conn.commit()

    async def async_find_bot(
        self,
        *,
        enterprise_id: Optional[str],
        team_id: Optional[str],
        is_enterprise_install: Optional[bool] = False,
    ) -> Optional[Bot]:
        return self.find_bot(
            enterprise_id=enterprise_id,
            team_id=team_id,
            is_enterprise_install=is_enterprise_install,
        )

    def find_bot(
        self,
        *,
        enterprise_id: Optional[str],
        team_id: Optional[str],
        is_enterprise_install: Optional[bool] = False,
    ) -> Optional[Bot]:
        if is_enterprise_install or team_id is None:
            team_id = ""

        try:
            with self.connect() as conn:
                cur = conn.execute(
                    """
                    select
                        app_id,
                        enterprise_id,
                        enterprise_name,
                        team_id,
                        team_name,
                        bot_token,
                        bot_id,
                        bot_user_id,
                        bot_scopes,
                        bot_refresh_token,  -- since v3.8
                        bot_token_expires_at,  -- since v3.8
                        is_enterprise_install,
                        installed_at
                    from
                        slack_bots
                    where
                        client_id = ?
                        and
                        enterprise_id = ?
                        and
                        team_id = ?
                    order by installed_at desc
                    limit 1
                    """,
                    [self.client_id, enterprise_id or "", team_id or ""],
                )
                row = cur.fetchone()
                result = "found" if row and len(row) > 0 else "not found"
                self.logger.debug(f"find_bot's query result: {result} (database: {self.database})")
                if row and len(row) > 0:
                    bot = Bot(
                        app_id=row[0],
                        enterprise_id=row[1],
                        enterprise_name=row[2],
                        team_id=row[3],
                        team_name=row[4],
                        bot_token=row[5],
                        bot_id=row[6],
                        bot_user_id=row[7],
                        bot_scopes=row[8],
                        bot_refresh_token=row[9],
                        bot_token_expires_at=row[10],
                        is_enterprise_install=row[11],
                        installed_at=row[12],
                    )
                    return bot
                return None

        except Exception as e:  # skipcq: PYL-W0703
            message = f"Failed to find bot installation data for enterprise: {enterprise_id}, team: {team_id}: {e}"
            if self.logger.level <= logging.DEBUG:
                self.logger.exception(message)
            else:
                self.logger.warning(message)
            return None

    async def async_find_installation(
        self,
        *,
        enterprise_id: Optional[str],
        team_id: Optional[str],
        user_id: Optional[str] = None,
        is_enterprise_install: Optional[bool] = False,
    ) -> Optional[Installation]:
        return self.find_installation(
            enterprise_id=enterprise_id,
            team_id=team_id,
            user_id=user_id,
            is_enterprise_install=is_enterprise_install,
        )

    def find_installation(
        self,
        *,
        enterprise_id: Optional[str],
        team_id: Optional[str],
        user_id: Optional[str] = None,
        is_enterprise_install: Optional[bool] = False,
    ) -> Optional[Installation]:
        if is_enterprise_install or team_id is None:
            team_id = ""

        try:
            with self.connect() as conn:
                row = None
                columns = """
                    app_id,
                    enterprise_id,
                    enterprise_name,
                    enterprise_url,
                    team_id,
                    team_name,
                    bot_token,
                    bot_id,
                    bot_user_id,
                    bot_scopes,
                    bot_refresh_token,  -- since v3.8
                    bot_token_expires_at,  -- since v3.8
                    user_id,
                    user_token,
                    user_scopes,
                    user_refresh_token,  -- since v3.8
                    user_token_expires_at,  -- since v3.8
                    incoming_webhook_url,
                    incoming_webhook_channel,
                    incoming_webhook_channel_id,
                    incoming_webhook_configuration_url,
                    is_enterprise_install,
                    token_type,
                    installed_at
                """
                if user_id is None:
                    cur = conn.execute(
                        f"""
                        select
                            {columns}
                        from
                            slack_installations
                        where
                            client_id = ?
                            and
                            enterprise_id = ?
                            and
                            team_id = ?
                        order by installed_at desc
                        limit 1
                        """,
                        [self.client_id, enterprise_id or "", team_id],
                    )
                    row = cur.fetchone()
                else:
                    cur = conn.execute(
                        f"""
                        select
                            {columns}
                        from
                            slack_installations
                        where
                            client_id = ?
                            and
                            enterprise_id = ?
                            and
                            team_id = ?
                            and
                            user_id = ?
                        order by installed_at desc
                        limit 1
                        """,
                        [self.client_id, enterprise_id or "", team_id, user_id],
                    )
                    row = cur.fetchone()

                if row is None:
                    return None

                result = "found" if row and len(row) > 0 else "not found"
                self.logger.debug(f"find_installation's query result: {result} (database: {self.database})")
                if row and len(row) > 0:
                    installation = Installation(
                        app_id=row[0],
                        enterprise_id=row[1],
                        enterprise_name=row[2],
                        enterprise_url=row[3],
                        team_id=row[4],
                        team_name=row[5],
                        bot_token=row[6],
                        bot_id=row[7],
                        bot_user_id=row[8],
                        bot_scopes=row[9],
                        bot_refresh_token=row[10],
                        bot_token_expires_at=row[11],
                        user_id=row[12],
                        user_token=row[13],
                        user_scopes=row[14],
                        user_refresh_token=row[15],
                        user_token_expires_at=row[16],
                        incoming_webhook_url=row[17],
                        incoming_webhook_channel=row[18],
                        incoming_webhook_channel_id=row[19],
                        incoming_webhook_configuration_url=row[20],
                        is_enterprise_install=row[21],
                        token_type=row[22],
                        installed_at=row[23],
                    )

                    if user_id is not None:
                        # Retrieve the latest bot token, just in case
                        # See also: https://github.com/slackapi/bolt-python/issues/664
                        cur = conn.execute(
                            """
                            select
                                bot_token,
                                bot_id,
                                bot_user_id,
                                bot_scopes,
                                bot_refresh_token,
                                bot_token_expires_at
                            from
                                slack_installations
                            where
                                client_id = ?
                                and
                                enterprise_id = ?
                                and
                                team_id = ?
                                and
                                bot_token is not null
                            order by installed_at desc
                            limit 1
                            """,
                            [self.client_id, enterprise_id or "", team_id],
                        )
                        row = cur.fetchone()
                        installation.bot_token = row[0]
                        installation.bot_id = row[1]
                        installation.bot_user_id = row[2]
                        installation.bot_scopes = row[3]
                        installation.bot_refresh_token = row[4]
                        installation.bot_token_expires_at = row[5]

                    return installation
                return None

        except Exception as e:  # skipcq: PYL-W0703
            message = f"Failed to find an installation data for enterprise: {enterprise_id}, team: {team_id}: {e}"
            if self.logger.level <= logging.DEBUG:
                self.logger.exception(message)
            else:
                self.logger.warning(message)
            return None

    def delete_bot(self, *, enterprise_id: Optional[str], team_id: Optional[str]) -> None:
        try:
            with self.connect() as conn:
                conn.execute(
                    """
                    delete
                    from
                        slack_bots
                    where
                        client_id = ?
                        and
                        enterprise_id = ?
                        and
                        team_id = ?
                    """,
                    [self.client_id, enterprise_id or "", team_id or ""],
                )
                conn.commit()
        except Exception as e:  # skipcq: PYL-W0703
            message = f"Failed to delete bot installation data for enterprise: {enterprise_id}, team: {team_id}: {e}"
            if self.logger.level <= logging.DEBUG:
                self.logger.exception(message)
            else:
                self.logger.warning(message)

    def delete_installation(
        self,
        *,
        enterprise_id: Optional[str],
        team_id: Optional[str],
        user_id: Optional[str] = None,
    ) -> None:
        try:
            with self.connect() as conn:
                if user_id is None:
                    conn.execute(
                        """
                        delete
                        from
                            slack_installations
                        where
                            client_id = ?
                            and
                            enterprise_id = ?
                            and
                            team_id = ?
                        """,
                        [self.client_id, enterprise_id or "", team_id],
                    )
                else:
                    conn.execute(
                        """
                        delete
                        from
                            slack_installations
                        where
                            client_id = ?
                            and
                            enterprise_id = ?
                            and
                            team_id = ?
                            and
                            user_id = ?
                        """,
                        [self.client_id, enterprise_id or "", team_id, user_id],
                    )
                conn.commit()
        except Exception as e:  # skipcq: PYL-W0703
            message = f"Failed to delete installation data for enterprise: {enterprise_id}, team: {team_id}: {e}"
            if self.logger.level <= logging.DEBUG:
                self.logger.exception(message)
            else:
                self.logger.warning(message)

Ancestors

Instance variables

prop logger : logging.Logger
Expand source code
@property
def logger(self) -> Logger:
    if self._logger is None:
        self._logger = logging.getLogger(__name__)
    return self._logger

Methods

def connect(self) ‑> sqlite3.Connection
def create_tables(self)
def init(self)

Inherited members