### Project Setup and Dependency Installation Source: https://docs.slack.dev/tools/python-slack-sdk/tutorial/uploading-files These commands create a new virtual environment, install the slack-sdk, and install other required library dependencies. ```bash echo 'slack-sdk>=3.12,<4' > requirements.txt python3 -m venv .venv source .venv/bin/activate pip install -U pip pip install -r requirements.txt ``` -------------------------------- ### Installation Store Methods Source: https://docs.slack.dev/tools/python-slack-sdk/reference/oauth/installation_store/index.html Example methods for managing installation data, including deleting by glob pattern and creating directories. ```python def _delete_by_glob(self, e_id: str, t_id: str, filepath_glob: str): for filepath in glob.glob(filepath_glob): try: os.remove(filepath) except FileNotFoundError as e: message = f"Failed to delete installation data for enterprise: {e_id}, team: {t_id}: {e}" self.logger.warning(message) @staticmethod def _mkdir(path: Union[str, Path]): if isinstance(path, str): path = Path(path) path.mkdir(parents=True, exist_ok=True) ``` -------------------------------- ### API Test with WebClient Source: https://docs.slack.dev/tools/python-slack-sdk/installation A Python REPL example to verify the installation by creating a `WebClient` instance and calling the `api_test` method. ```python import os import logging from slack_sdk import WebClient logging.basicConfig(level=logging.DEBUG) client = WebClient(token=os.environ["SLACK_BOT_TOKEN"]) res = client.api_test() ``` -------------------------------- ### Virtual Environment Setup and SDK Installation Source: https://docs.slack.dev/tools/python-slack-sdk/installation Commands to create a Python virtual environment, activate it, install the `slack_sdk` package, and set the bot token environment variable. ```bash # Create a dedicated virtual env for running your Python scriptspython -m venv .venv# Run .venv\Scripts\activate on Windows OSsource .venv/bin/activate# Install slack_sdk PyPI packagepip install "slack_sdk>=3.0"# Set your token as an env variable (`set` command for Windows OS)export SLACK_BOT_TOKEN=xoxb-*** ``` -------------------------------- ### Install from source Source: https://docs.slack.dev/tools/python-slack-sdk/installation Cloning the repository and installing the SDK locally for development. ```bash git clone https://github.com/slackapi/python-slack-sdk.git cd python-slack-sdk python3 -m venv .venv source .venv/bin/activate pip install -U pippip install -e . # install the SDK project into the virtual env ``` -------------------------------- ### Install using pip Source: https://docs.slack.dev/tools/python-slack-sdk/installation The recommended way to install the slack-sdk package using pip. ```bash pip install slack-sdk ``` -------------------------------- ### Installation Class Definition Source: https://docs.slack.dev/tools/python-slack-sdk/reference/oauth/installation_store/models/installation.html This code snippet shows the definition of the Installation class, including its attributes and their types. ```python class Installation: app_id: Optional[str] enterprise_id: Optional[str] enterprise_name: Optional[str] enterprise_url: Optional[str] team_id: Optional[str] team_name: Optional[str] bot_token: Optional[str] bot_id: Optional[str] bot_user_id: Optional[str] bot_scopes: Optional[Sequence[str]] bot_refresh_token: Optional[str] # only when token rotation is enabled # only when token rotation is enabled # Unix time (seconds): only when token rotation is enabled bot_token_expires_at: Optional[int] user_id: str user_token: Optional[str] user_scopes: Optional[Sequence[str]] user_refresh_token: Optional[str] # only when token rotation is enabled # Unix time (seconds): only when token rotation is enabled user_token_expires_at: Optional[int] incoming_webhook_url: Optional[str] incoming_webhook_channel: Optional[str] incoming_webhook_channel_id: Optional[str] incoming_webhook_configuration_url: Optional[str] is_enterprise_install: bool token_type: Optional[str] installed_at: float custom_values: Dict[str, Any] def __init__( self, *, app_id: Optional[str] = None, # org / workspace enterprise_id: Optional[str] = None, enterprise_name: Optional[str] = None, enterprise_url: Optional[str] = None, team_id: Optional[str] = None, team_name: Optional[str] = None, # bot bot_token: Optional[str] = None, bot_id: Optional[str] = None, bot_user_id: Optional[str] = None, bot_scopes: Union[str, Sequence[str]] = "", bot_refresh_token: Optional[str] = None, # only when token rotation is enabled # only when token rotation is enabled bot_token_expires_in: Optional[int] = None, # only for duplicating this object # only when token rotation is enabled bot_token_expires_at: Optional[Union[int, datetime, str]] = None, # installer user_id: str, user_token: Optional[str] = None, user_scopes: Union[str, Sequence[str]] = "", user_refresh_token: Optional[str] = None, # only when token rotation is enabled # only when token rotation is enabled user_token_expires_in: Optional[int] = None, # only for duplicating this object # only when token rotation is enabled user_token_expires_at: Optional[Union[int, datetime, str]] = None, # incoming webhook incoming_webhook_url: Optional[str] = None, incoming_webhook_channel: Optional[str] = None, incoming_webhook_channel_id: Optional[str] = None, incoming_webhook_configuration_url: Optional[str] = None, # org app is_enterprise_install: Optional[bool] = False, token_type: Optional[str] = None, # timestamps # The expected value type is float but the internals handle other types too # for str values, we supports only ISO datetime format. installed_at: Optional[Union[float, datetime, str]] = None, # custom values custom_values: Optional[Dict[str, Any]] = None, ): self.app_id = app_id self.enterprise_id = enterprise_id self.enterprise_name = enterprise_name self.enterprise_url = enterprise_url self.team_id = team_id self.team_name = team_name self.bot_token = bot_token self.bot_id = bot_id self.bot_user_id = bot_user_id if isinstance(bot_scopes, str): self.bot_scopes = bot_scopes.split(",") if len(bot_scopes) > 0 else [] else: self.bot_scopes = bot_scopes self.bot_refresh_token = bot_refresh_token if bot_token_expires_at is not None: ``` -------------------------------- ### Installation Store Class Definition Source: https://docs.slack.dev/tools/python-slack-sdk/reference/oauth/installation_store/index.html This Python code defines the InstallationStore class, which holds installation data for a Slack app. It includes methods for initializing the store, converting it to a Bot object, setting and getting custom values, and converting the store to a dictionary. ```python class InstallationStore: def __init__( self, app_id: Optional[str] = None, enterprise_id: Optional[str] = None, enterprise_name: Optional[str] = None, enterprise_url: Optional[str] = None, team_id: Optional[str] = None, team_name: Optional[str] = None, bot_token: Optional[str] = None, bot_id: Optional[str] = None, bot_user_id: Optional[str] = None, bot_scopes: Optional[List[str]] = None, bot_refresh_token: Optional[str] = None, bot_token_expires_in: Optional[int] = None, bot_token_expires_at: Optional[int] = None, user_id: Optional[str] = None, user_token: Optional[str] = None, user_scopes: Optional[List[str]] = None, user_refresh_token: Optional[str] = None, user_token_expires_in: Optional[int] = None, user_token_expires_at: Optional[int] = None, incoming_webhook_url: Optional[str] = None, incoming_webhook_channel: Optional[str] = None, incoming_webhook_channel_id: Optional[str] = None, incoming_webhook_configuration_url: Optional[str] = None, is_enterprise_install: Optional[bool] = None, token_type: Optional[str] = None, installed_at: Optional[float] = None, custom_values: Optional[Dict[str, Any]] = None, ): self.app_id = app_id self.enterprise_id = enterprise_id self.enterprise_name = enterprise_name self.enterprise_url = enterprise_url self.team_id = team_id self.team_name = team_name self.bot_token = bot_token self.bot_id = bot_id self.bot_user_id = bot_user_id if isinstance(bot_scopes, str): self.bot_scopes = bot_scopes.split(",") if len(bot_scopes) > 0 else [] else: self.bot_scopes = bot_scopes self.bot_refresh_token = bot_refresh_token if bot_token_expires_at is not None: self.bot_token_expires_at = _timestamp_to_type(bot_token_expires_at, int) elif bot_token_expires_in is not None: self.bot_token_expires_at = int(time()) + bot_token_expires_in else: self.bot_token_expires_at = None self.user_id = user_id self.user_token = user_token if isinstance(user_scopes, str): self.user_scopes = user_scopes.split(",") if len(user_scopes) > 0 else [] else: self.user_scopes = user_scopes self.user_refresh_token = user_refresh_token if user_token_expires_at is not None: self.user_token_expires_at = _timestamp_to_type(user_token_expires_at, int) elif user_token_expires_in is not None: self.user_token_expires_at = int(time()) + user_token_expires_in else: self.user_token_expires_at = None self.incoming_webhook_url = incoming_webhook_url self.incoming_webhook_channel = incoming_webhook_channel self.incoming_webhook_channel_id = incoming_webhook_channel_id self.incoming_webhook_configuration_url = incoming_webhook_configuration_url self.is_enterprise_install = is_enterprise_install or False self.token_type = token_type if installed_at is None: self.installed_at = datetime.now().timestamp() else: self.installed_at = _timestamp_to_type(installed_at, float) self.custom_values = custom_values if custom_values is not None else {} def to_bot(self) -> Bot: return Bot( app_id=self.app_id, enterprise_id=self.enterprise_id, enterprise_name=self.enterprise_name, team_id=self.team_id, team_name=self.team_name, bot_token=self.bot_token, # type: ignore[arg-type] bot_id=self.bot_id, # type: ignore[arg-type] bot_user_id=self.bot_user_id, # type: ignore[arg-type] bot_scopes=self.bot_scopes, # type: ignore[arg-type] bot_refresh_token=self.bot_refresh_token, bot_token_expires_at=self.bot_token_expires_at, is_enterprise_install=self.is_enterprise_install, installed_at=self.installed_at, custom_values=self.custom_values, ) def set_custom_value(self, name: str, value: Any): self.custom_values[name] = value def get_custom_value(self, name: str) -> Optional[Any]: return self.custom_values.get(name) def _to_standard_value_dict(self) -> Dict[str, Any]: return { "app_id": self.app_id, "enterprise_id": self.enterprise_id, "enterprise_name": self.enterprise_name, "enterprise_url": self.enterprise_url, "team_id": self.team_id, "team_name": self.team_name, "bot_token": self.bot_token, "bot_id": self.bot_id, "bot_user_id": self.bot_user_id, "bot_scopes": ",".join(self.bot_scopes) if self.bot_scopes else None, "bot_refresh_token": self.bot_refresh_token, "bot_token_expires_at": ( datetime.fromtimestamp(self.bot_token_expires_at, tz=timezone.utc) if self.bot_token_expires_at is not None else None ), "user_id": self.user_id, "user_token": self.user_token, "user_scopes": ",".join(self.user_scopes) if self.user_scopes else None, "user_refresh_token": self.user_refresh_token, "user_token_expires_at": ( datetime.fromtimestamp(self.user_token_expires_at, tz=timezone.utc) if self.user_token_expires_at is not None else None ), "incoming_webhook_url": self.incoming_webhook_url, "incoming_webhook_channel": self.incoming_webhook_channel, "incoming_webhook_channel_id": self.incoming_webhook_channel_id, "incoming_webhook_configuration_url": self.incoming_webhook_configuration_url, "is_enterprise_install": self.is_enterprise_install, "token_type": self.token_type, "installed_at": datetime.fromtimestamp(self.installed_at, tz=timezone.utc), } ``` -------------------------------- ### dialog_open example Source: https://docs.slack.dev/tools/python-slack-sdk/reference/web/legacy_client.html Example of how to call the dialog_open method. ```python kwargs.update({"dialog": dialog, "trigger_id": trigger_id}) kwargs = _remove_none_values(kwargs) # NOTE: As the dialog can be a dict, this API call works only with json format. return self.api_call("dialog.open", json=kwargs) ``` -------------------------------- ### dnd_info example Source: https://docs.slack.dev/tools/python-slack-sdk/reference/web/legacy_client.html Example of how to call the dnd_info method. ```python kwargs.update({"team_id": team_id, "user": user}) return self.api_call("dnd.info", http_verb="GET", params=kwargs) ``` -------------------------------- ### to_bot Method Source: https://docs.slack.dev/tools/python-slack-sdk/reference/oauth/installation_store/models/installation.html Converts the current Installation object into a Bot object. ```python def to_bot(self) -> Bot: return Bot( app_id=self.app_id, enterprise_id=self.enterprise_id, enterprise_name=self.enterprise_name, team_id=self.team_id, team_name=self.team_name, bot_token=self.bot_token, # type: ignore[arg-type] bot_id=self.bot_id, # type: ignore[arg-type] bot_user_id=self.bot_user_id, # type: ignore[arg-type] bot_scopes=self.bot_scopes, # type: ignore[arg-type] bot_refresh_token=self.bot_refresh_token, bot_token_expires_at=self.bot_token_expires_at, is_enterprise_install=self.is_enterprise_install, installed_at=self.installed_at, custom_values=self.custom_values, ) ``` -------------------------------- ### Set environment variable for token Source: https://docs.slack.dev/tools/python-slack-sdk/installation Example of setting the SLACK_BOT_TOKEN environment variable before running an application. ```bash SLACK_BOT_TOKEN="xoxb-111-222-xxxxx" python myapp.py ``` -------------------------------- ### File Installation Store Implementation Source: https://docs.slack.dev/tools/python-slack-sdk/reference/oauth/installation_store/file/index.html This code snippet shows the implementation of the File Installation Store, including methods for saving, finding, and deleting installation data. ```python 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]: none = "none" e_id = enterprise_id or none t_id = team_id or none if is_enterprise_install: t_id = none installation_filepath = f"{self.base_dir}/{e_id}-{t_id}/installer-latest" if user_id is not None: installation_filepath = f"{self.base_dir}/{e_id}-{t_id}/installer-{user_id}-latest" try: installation: Optional[Installation] = None with open(installation_filepath) as f: data = json.loads(f.read()) installation = Installation(**data) has_user_installation = user_id is not None and installation is not None no_bot_token_installation = installation is not None and installation.bot_token is None should_find_bot_installation = has_user_installation or no_bot_token_installation if should_find_bot_installation: # Retrieve the latest bot token, just in case # See also: https://github.com/slackapi/bolt-python/issues/664 latest_bot_installation = self.find_bot( enterprise_id=enterprise_id, team_id=team_id, is_enterprise_install=is_enterprise_install, ) if latest_bot_installation is not None and installation.bot_token != latest_bot_installation.bot_token: # NOTE: this logic is based on the assumption that every single installation has bot scopes # If you need to installation patterns without bot scopes in the same S3 bucket, # please fork this code and implement your own logic. installation.bot_id = latest_bot_installation.bot_id installation.bot_user_id = latest_bot_installation.bot_user_id installation.bot_token = latest_bot_installation.bot_token installation.bot_scopes = latest_bot_installation.bot_scopes installation.bot_refresh_token = latest_bot_installation.bot_refresh_token installation.bot_token_expires_at = latest_bot_installation.bot_token_expires_at return installation except FileNotFoundError as e: message = f"Installation data missing for enterprise: {e_id}, team: {t_id}: {e}" self.logger.debug(message) return None async def async_delete_bot(self, *, enterprise_id: Optional[str], team_id: Optional[str]) -> None: return self.delete_bot(enterprise_id=enterprise_id, team_id=team_id) def delete_bot(self, *, enterprise_id: Optional[str], team_id: Optional[str]) -> None: none = "none" e_id = enterprise_id or none t_id = team_id or none filepath_glob = f"{self.base_dir}/{e_id}-{t_id}/bot-*" self._delete_by_glob(e_id, t_id, filepath_glob) async def async_delete_installation( self, *, enterprise_id: Optional[str], team_id: Optional[str], user_id: Optional[str] = None, ) -> None: return self.delete_installation(enterprise_id=enterprise_id, team_id=team_id, user_id=user_id) def delete_installation( self, *, enterprise_id: Optional[str], team_id: Optional[str], user_id: Optional[str] = None, ) -> None: none = "none" e_id = enterprise_id or none t_id = team_id or none if user_id is not None: filepath_glob = f"{self.base_dir}/{e_id}-{t_id}/installer-{user_id}-*" else: filepath_glob = f"{self.base_dir}/{e_id}-{t_id}/installer-*" self._delete_by_glob(e_id, t_id, filepath_glob) def _delete_by_glob(self, e_id: str, t_id: str, filepath_glob: str): for filepath in glob.glob(filepath_glob): try: os.remove(filepath) except FileNotFoundError as e: message = f"Failed to delete installation data for enterprise: {e_id}, team: {t_id}: {e}" self.logger.warning(message) @staticmethod def _mkdir(path: Union[str, Path]): if isinstance(path, str): path = Path(path) path.mkdir(parents=True, exist_ok=True) ``` -------------------------------- ### InstallationStore Installation Models Source: https://docs.slack.dev/tools/python-slack-sdk/reference/oauth/installation_store/models/index.html This code snippet shows the InstallationStore Installation models. ```python "token_type": self.token_type, "installed_at": datetime.fromtimestamp(self.installed_at, tz=timezone.utc), } def to_dict_for_copying(self) -> Dict[str, Any]: return {"custom_values": self.custom_values, **self._to_standard_value_dict()} def to_dict(self) -> Dict[str, Any]: # prioritize standard_values over custom_values # when the same keys exist in both return {**self.custom_values, **self._to_standard_value_dict()} ``` ``` -------------------------------- ### Find Installation Source: https://docs.slack.dev/tools/python-slack-sdk/reference/oauth/installation_store/index.html This method finds and returns an installation from the store. It supports finding installations by user ID and includes logic to retrieve the latest bot token if necessary, ensuring consistency. ```python 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]: none = "none" e_id = enterprise_id or none t_id = team_id or none if is_enterprise_install: t_id = none installation_filepath = f"{self.base_dir}/{e_id}-{t_id}/installer-latest" if user_id is not None: installation_filepath = f"{self.base_dir}/{e_id}-{t_id}/installer-{user_id}-latest" try: installation: Optional[Installation] = None with open(installation_filepath) as f: data = json.loads(f.read()) installation = Installation(**data) has_user_installation = user_id is not None and installation is not None no_bot_token_installation = installation is not None and installation.bot_token is None should_find_bot_installation = has_user_installation or no_bot_token_installation if should_find_bot_installation: # Retrieve the latest bot token, just in case # See also: https://github.com/slackapi/bolt-python/issues/664 latest_bot_installation = self.find_bot( enterprise_id=enterprise_id, team_id=team_id, is_enterprise_install=is_enterprise_install, ) if latest_bot_installation is not None and installation.bot_token != latest_bot_installation.bot_token: # NOTE: this logic is based on the assumption that every single installation has bot scopes # If you need to installation patterns without bot scopes in the same S3 bucket, # please fork this code and implement your own logic. installation.bot_id = latest_bot_installation.bot_id installation.bot_user_id = latest_bot_installation.bot_user_id installation.bot_token = latest_bot_installation.bot_token installation.bot_scopes = latest_bot_installation.bot_scopes installation.bot_refresh_token = latest_bot_installation.bot_refresh_token installation.bot_token_expires_at = latest_bot_installation.bot_token_expires_at return installation except FileNotFoundError as e: message = f"Installation data missing for enterprise: {e_id}, team: {t_id}: {e}" self.logger.debug(message) return None ``` -------------------------------- ### conversations.create example Source: https://docs.slack.dev/tools/python-slack-sdk/reference/web/client.html Example of initiating a public or private channel-based conversation. ```python kwargs.update({"name": name, "is_private": is_private, "team_id": team_id}) return self.api_call("conversations.create", params=kwargs) ``` -------------------------------- ### dnd_teamInfo example Source: https://docs.slack.dev/tools/python-slack-sdk/reference/web/legacy_client.html Example of how to call the dnd_teamInfo method. ```python kwargs.update({"users": users}) kwargs.update({"team_id": team_id}) return self.api_call("dnd.teamInfo", http_verb="GET", params=kwargs) ``` -------------------------------- ### migration.exchange example Source: https://docs.slack.dev/tools/python-slack-sdk/reference Example of how to use the migration.exchange API method. ```python kwargs.update({"users": users}) kwargs.update({"team_id": team_id, "to_old": to_old}) return self.api_call("migration.exchange", http_verb="GET", params=kwargs) ``` -------------------------------- ### SQLAlchemy Installation Store Methods Source: https://docs.slack.dev/tools/python-slack-sdk/reference/oauth/installation_store/sqlalchemy/index.html This snippet shows the implementation of `find_bot`, `delete_bot`, `delete_installation`, `build_installation_entity`, and `build_bot_entity` within the SQLAlchemy Installation Store. ```python no_bot_token_installation = installation is not None and installation.bot_token is None should_find_bot_installation = has_user_installation or no_bot_token_installation if should_find_bot_installation: # Retrieve the latest bot token, just in case # See also: https://github.com/slackapi/bolt-python/issues/664 latest_bot_installation = self.find_bot( enterprise_id=enterprise_id, team_id=team_id, is_enterprise_install=is_enterprise_install, ) if ( latest_bot_installation is not None and installation is not None and installation.bot_token != latest_bot_installation.bot_token ): installation.bot_id = latest_bot_installation.bot_id installation.bot_user_id = latest_bot_installation.bot_user_id installation.bot_token = latest_bot_installation.bot_token installation.bot_scopes = latest_bot_installation.bot_scopes installation.bot_refresh_token = latest_bot_installation.bot_refresh_token installation.bot_token_expires_at = latest_bot_installation.bot_token_expires_at return installation def delete_bot(self, *, enterprise_id: Optional[str], team_id: Optional[str]) -> None: table = self.bots c = table.c with self.engine.begin() as conn: deletion = table.delete().where( and_( c.client_id == self.client_id, c.enterprise_id == enterprise_id, c.team_id == team_id, ) ) conn.execute(deletion) def delete_installation( self, *, enterprise_id: Optional[str], team_id: Optional[str], user_id: Optional[str] = None, ) -> None: table = self.installations c = table.c with self.engine.begin() as conn: if user_id is not None: deletion = table.delete().where( and_( c.client_id == self.client_id, c.enterprise_id == enterprise_id, c.team_id == team_id, c.user_id == user_id, ) ) conn.execute(deletion) else: deletion = table.delete().where( and_( c.client_id == self.client_id, c.enterprise_id == enterprise_id, c.team_id == team_id, ) ) conn.execute(deletion) @classmethod def build_installation_entity(cls, row) -> Installation: return Installation( app_id=row["app_id"], enterprise_id=row["enterprise_id"], enterprise_name=row["enterprise_name"], enterprise_url=row["enterprise_url"], team_id=row["team_id"], team_name=row["team_name"], bot_token=row["bot_token"], bot_id=row["bot_id"], bot_user_id=row["bot_user_id"], bot_scopes=row["bot_scopes"], bot_refresh_token=row["bot_refresh_token"], bot_token_expires_at=row["bot_token_expires_at"], user_id=row["user_id"], user_token=row["user_token"], user_scopes=row["user_scopes"], user_refresh_token=row["user_refresh_token"], user_token_expires_at=row["user_token_expires_at"], # Only the incoming webhook issued in the latest installation is set in this logic incoming_webhook_url=row["incoming_webhook_url"], incoming_webhook_channel=row["incoming_webhook_channel"], incoming_webhook_channel_id=row["incoming_webhook_channel_id"], incoming_webhook_configuration_url=row["incoming_webhook_configuration_url"], is_enterprise_install=row["is_enterprise_install"], token_type=row["token_type"], installed_at=row["installed_at"], ) @classmethod def build_bot_entity(cls, row) -> Bot: return Bot( app_id=row["app_id"], enterprise_id=row["enterprise_id"], enterprise_name=row["enterprise_name"], team_id=row["team_id"], team_name=row["team_name"], bot_token=row["bot_token"], bot_id=row["bot_id"], bot_user_id=row["bot_user_id"], bot_scopes=row["bot_scopes"], bot_refresh_token=row["bot_refresh_token"], bot_token_expires_at=row["bot_token_expires_at"], is_enterprise_install=row["is_enterprise_install"], installed_at=row["installed_at"], ) ``` -------------------------------- ### conversations.create example Source: https://docs.slack.dev/tools/python-slack-sdk/reference/index.html Example of initiating a public or private channel-based conversation. ```python kwargs.update({"name": name, "is_private": is_private, "team_id": team_id}) return self.api_call("conversations.create", params=kwargs) ``` -------------------------------- ### SQLAlchemy Installation Store - find_installation method Source: https://docs.slack.dev/tools/python-slack-sdk/reference/oauth/installation_store/sqlalchemy/index.html The `find_installation` method retrieves the latest installation data for a given enterprise, team, and optionally user. ```python 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 = None c = self.installations.c where_clause = and_( c.client_id == self.client_id, c.enterprise_id == enterprise_id, c.team_id == team_id, ) if user_id is not None: where_clause = and_( c.client_id == self.client_id, c.enterprise_id == enterprise_id, c.team_id == team_id, c.user_id == user_id, ) query = self.installations.select().where(where_clause).order_by(desc(c.installed_at)).limit(1) installation: Optional[Installation] = None with self.engine.connect() as conn: result: object = conn.execute(query) for row in result.mappings(): # type: ignore[attr-defined] installation = self.build_installation_entity(row) has_user_installation = user_id is not None and installation is not None ``` -------------------------------- ### Find Bot Installation Source: https://docs.slack.dev/tools/python-slack-sdk/reference/oauth/installation_store/index.html This method finds and returns a bot installation from the store. It handles cases where enterprise or team IDs might be missing and attempts to load installation data from a file. ```python def find_bot(self, *, enterprise_id: Optional[str], team_id: Optional[str], is_enterprise_install: Optional[bool] = False) -> Optional[Bot]: none = "none" e_id = enterprise_id or none t_id = team_id or none if is_enterprise_install: t_id = none bot_filepath = f"{self.base_dir}/{e_id}-{t_id}/bot-latest" try: with open(bot_filepath) as f: data = json.loads(f.read()) return Bot(**data) except FileNotFoundError as e: message = f"Installation data missing for enterprise: {e_id}, team: {t_id}: {e}" self.logger.debug(message) return None ``` -------------------------------- ### team.info Source: https://docs.slack.dev/tools/python-slack-sdk/reference/index.html Example of how to get information about the current team. ```python def team_info( self, *, team: Optional[str] = None, domain: Optional[str] = None, **kwargs, ) -> SlackResponse: """Gets information about the current team. https://docs.slack.dev/reference/methods/team.info """ kwargs.update({"team": team, "domain": domain}) return self.api_call("team.info", http_verb="GET", params=kwargs) ``` -------------------------------- ### team.integrationLogs Source: https://docs.slack.dev/tools/python-slack-sdk/reference/index.html Example of how to get integration logs for the current team. ```python def team_integrationLogs( self, *, app_id: Optional[str] = None, change_type: Optional[str] = None, count: Optional[Union[int, str]] = None, page: Optional[Union[int, str]] = None, service_id: Optional[str] = None, team_id: Optional[str] = None, user: Optional[str] = None, **kwargs, ) -> SlackResponse: """Gets the integration logs for the current team. https://docs.slack.dev/reference/methods/team.integrationLogs """ kwargs.update( { "app_id": app_id, "change_type": change_type, "count": count, "page": page, "service_id": service_id, "team_id": team_id, "user": user, } ) return self.api_call("team.integrationLogs", http_verb="GET", params=kwargs) ``` -------------------------------- ### rtm.connect Source: https://docs.slack.dev/tools/python-slack-sdk/reference/web/index.html Starts a Real Time Messaging session (sent as a GET request). ```python kwargs.update({"batch_presence_aware": batch_presence_aware, "presence_sub": presence_sub}) return self.api_call("rtm.connect", http_verb="GET", params=kwargs) ``` -------------------------------- ### FileInstallationStore Constructor Source: https://docs.slack.dev/tools/python-slack-sdk/reference/oauth/installation_store/index.html Initializes the FileInstallationStore with optional parameters for base directory, historical data enablement, client ID, and logger. ```python class FileInstallationStore(InstallationStore, AsyncInstallationStore): def __init__( self, *, base_dir: str = str(Path.home()) + "/.bolt-app-installation", historical_data_enabled: bool = True, client_id: Optional[str] = None, logger: Logger = logging.getLogger(__name__), ): self.base_dir = base_dir self.historical_data_enabled = historical_data_enabled self.client_id = client_id if self.client_id is not None: self.base_dir = f"{self.base_dir}/{self.client_id}" self._logger = logger ``` -------------------------------- ### Bots Info Method Source: https://docs.slack.dev/tools/python-slack-sdk/reference/web/legacy_client.html Example of how to get bot user information using the bots.info method. ```python def bots_info( self, *, bot: Optional[str] = None, team_id: Optional[str] = None, **kwargs, ) -> Union[Future, SlackResponse]: """Gets information about a bot user. https://docs.slack.dev/reference/methods/bots.info """ kwargs.update({"bot": bot, "team_id": team_id}) return self.api_call("bots.info", http_verb="GET", params=kwargs) ``` -------------------------------- ### apps_event_authorizations_list Source: https://docs.slack.dev/tools/python-slack-sdk/reference/index.html Get a list of authorizations for the given event context. Each authorization represents an app installation that the event is visible to. ```python def apps_event_authorizations_list( self, *, event_context: str, cursor: Optional[str] = None, limit: Optional[int] = None, **kwargs, ) -> SlackResponse: """Get a list of authorizations for the given event context. Each authorization represents an app installation that the event is visible to. https://docs.slack.dev/reference/methods/apps.event.authorizations.list """ kwargs.update({"event_context": event_context, "cursor": cursor, "limit": limit}) return self.api_call("apps.event.authorizations.list", params=kwargs) ``` -------------------------------- ### Installation Model Attributes and Methods Source: https://docs.slack.dev/tools/python-slack-sdk/reference/oauth/installation_store/models/installation.html This Python code snippet shows the attributes and methods of the Installation model, including how it handles token expiration, scopes, and custom values, and how it can be converted to other formats like Bot or dictionary representations. ```python from datetime import datetime, timezone from time import time from typing import Any, Dict, Optional from .bot import Bot def _timestamp_to_type(timestamp: Optional[float], type_hint: Any) -> Any: if timestamp is None: return None if type_hint == int: return int(timestamp) elif type_hint == float: return float(timestamp) return timestamp class Installation: def __init__( self, app_id: Optional[str] = None, enterprise_id: Optional[str] = None, enterprise_name: Optional[str] = None, enterprise_url: Optional[str] = None, team_id: Optional[str] = None, team_name: Optional[str] = None, bot_token: Optional[str] = None, bot_id: Optional[str] = None, bot_user_id: Optional[str] = None, bot_scopes: Optional[list[str]] = None, bot_refresh_token: Optional[str] = None, bot_token_expires_in: Optional[int] = None, bot_token_expires_at: Optional[float] = None, user_id: Optional[str] = None, user_token: Optional[str] = None, user_scopes: Optional[list[str]] = None, user_refresh_token: Optional[str] = None, user_token_expires_in: Optional[int] = None, user_token_expires_at: Optional[float] = None, incoming_webhook_url: Optional[str] = None, incoming_webhook_channel: Optional[str] = None, incoming_webhook_channel_id: Optional[str] = None, incoming_webhook_configuration_url: Optional[str] = None, is_enterprise_install: Optional[bool] = None, token_type: Optional[str] = None, installed_at: Optional[float] = None, custom_values: Optional[Dict[str, Any]] = None, ): self.app_id = app_id self.enterprise_id = enterprise_id self.enterprise_name = enterprise_name self.enterprise_url = enterprise_url self.team_id = team_id self.team_name = team_name self.bot_token = bot_token self.bot_id = bot_id self.bot_user_id = bot_user_id self.bot_scopes = bot_scopes self.bot_refresh_token = bot_refresh_token if bot_token_expires_at is not None: self.bot_token_expires_at = _timestamp_to_type(bot_token_expires_at, int) elif bot_token_expires_in is not None: self.bot_token_expires_at = int(time()) + bot_token_expires_in else: self.bot_token_expires_at = None self.user_id = user_id self.user_token = user_token if isinstance(user_scopes, str): self.user_scopes = user_scopes.split(",") if len(user_scopes) > 0 else [] else: self.user_scopes = user_scopes self.user_refresh_token = user_refresh_token if user_token_expires_at is not None: self.user_token_expires_at = _timestamp_to_type(user_token_expires_at, int) elif user_token_expires_in is not None: self.user_token_expires_at = int(time()) + user_token_expires_in else: self.user_token_expires_at = None self.incoming_webhook_url = incoming_webhook_url self.incoming_webhook_channel = incoming_webhook_channel self.incoming_webhook_channel_id = incoming_webhook_channel_id self.incoming_webhook_configuration_url = incoming_webhook_configuration_url self.is_enterprise_install = is_enterprise_install or False self.token_type = token_type if installed_at is None: self.installed_at = datetime.now().timestamp() else: self.installed_at = _timestamp_to_type(installed_at, float) self.custom_values = custom_values if custom_values is not None else {} def to_bot(self) -> Bot: return Bot( app_id=self.app_id, enterprise_id=self.enterprise_id, enterprise_name=self.enterprise_name, team_id=self.team_id, team_name=self.team_name, bot_token=self.bot_token, # type: ignore[arg-type] bot_id=self.bot_id, # type: ignore[arg-type] bot_user_id=self.bot_user_id, # type: ignore[arg-type] bot_scopes=self.bot_scopes, # type: ignore[arg-type] bot_refresh_token=self.bot_refresh_token, bot_token_expires_at=self.bot_token_expires_at, is_enterprise_install=self.is_enterprise_install, installed_at=self.installed_at, custom_values=self.custom_values, ) def set_custom_value(self, name: str, value: Any): self.custom_values[name] = value def get_custom_value(self, name: str) -> Optional[Any]: return self.custom_values.get(name) def _to_standard_value_dict(self) -> Dict[str, Any]: return { "app_id": self.app_id, "enterprise_id": self.enterprise_id, "enterprise_name": self.enterprise_name, "enterprise_url": self.enterprise_url, "team_id": self.team_id, "team_name": self.team_name, "bot_token": self.bot_token, "bot_id": self.bot_id, "bot_user_id": self.bot_user_id, "bot_scopes": ",".join(self.bot_scopes) if self.bot_scopes else None, "bot_refresh_token": self.bot_refresh_token, "bot_token_expires_at": ( datetime.fromtimestamp(self.bot_token_expires_at, tz=timezone.utc) if self.bot_token_expires_at is not None else None ), "user_id": self.user_id, "user_token": self.user_token, "user_scopes": ",".join(self.user_scopes) if self.user_scopes else None, "user_refresh_token": self.user_refresh_token, "user_token_expires_at": ( datetime.fromtimestamp(self.user_token_expires_at, tz=timezone.utc) if self.user_token_expires_at is not None else None ), "incoming_webhook_url": self.incoming_webhook_url, "incoming_webhook_channel": self.incoming_webhook_channel, "incoming_webhook_channel_id": self.incoming_webhook_channel_id, "incoming_webhook_configuration_url": self.incoming_webhook_configuration_url, "is_enterprise_install": self.is_enterprise_install, "token_type": self.token_type, "installed_at": datetime.fromtimestamp(self.installed_at, tz=timezone.utc), } def to_dict_for_copying(self) -> Dict[str, Any]: return {"custom_values": self.custom_values, **self._to_standard_value_dict()} def to_dict(self) -> Dict[str, Any]: # prioritize standard_values over custom_values # when the same keys exist in both return {**self.custom_values, **self._to_standard_value_dict()} ``` -------------------------------- ### apps.event.authorizations.list Source: https://docs.slack.dev/tools/python-slack-sdk/reference/index.html Get a list of authorizations for the given event context. Each authorization represents an app installation that the event is visible to. ```python """Get a list of authorizations for the given event context. Each authorization represents an app installation that the event is visible to. https://docs.slack.dev/reference/methods/apps.event.authorizations.list """ kwargs.update({"event_context": event_context, "cursor": cursor, "limit": limit}) return self.api_call("apps.event.authorizations.list", params=kwargs) ``` -------------------------------- ### Run test script Source: https://docs.slack.dev/tools/python-slack-sdk/installation Command to execute the test.py script. ```bash python test.py ``` -------------------------------- ### init Source: https://docs.slack.dev/tools/python-slack-sdk/reference/oauth/installation_store/sqlite3/index.html Initializes the SQLite3InstallationStore by checking if tables exist and creating them if necessary. ```python 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: self.create_tables() self.init_called = True ``` -------------------------------- ### Bots Info Method Source: https://docs.slack.dev/tools/python-slack-sdk/reference/web/async_client.html Example of how to get information about a bot user using the async client. ```python async def bots_info( self, *, bot: Optional[str] = None, team_id: Optional[str] = None, **kwargs, ) -> AsyncSlackResponse: """Gets information about a bot user. https://docs.slack.dev/reference/methods/bots.info """ kwargs.update({"bot": bot, "team_id": team_id}) return await self.api_call("bots.info", http_verb="GET", params=kwargs) ``` -------------------------------- ### Installation Store Methods Source: https://docs.slack.dev/tools/python-slack-sdk/reference/oauth/installation_store/index.html Methods for converting installation data to dictionaries, prioritizing standard values over custom values when keys overlap. ```python def to_dict_for_copying(self) -> Dict[str, Any]: return {"custom_values": self.custom_values, **self._to_standard_value_dict()} def to_dict(self) -> Dict[str, Any]: # prioritize standard_values over custom_values # when the same keys exist in both return {**self.custom_values, **self._to_standard_value_dict()} ``` -------------------------------- ### save Method Source: https://docs.slack.dev/tools/python-slack-sdk/reference/oauth/installation_store/index.html Saves installation data to the file system. It handles creating directories and writing installation details, with options for historical data. ```python def save(self, installation: Installation): none = "none" e_id = installation.enterprise_id or none t_id = installation.team_id or none team_installation_dir = f"{self.base_dir}/{e_id}-{t_id}" self._mkdir(team_installation_dir) self.save_bot(installation.to_bot()) if self.historical_data_enabled: history_version: str = str(installation.installed_at) # per workspace entity: str = json.dumps(installation.__dict__) with open(f"{team_installation_dir}/installer-latest", "w") as f: f.write(entity) with open(f"{team_installation_dir}/installer-{history_version}", "w") as f: f.write(entity) # per workspace per user u_id = installation.user_id or none entity = json.dumps(installation.__dict__) with open(f"{team_installation_dir}/installer-{u_id}-latest", "w") as f: f.write(entity) with open(f"{team_installation_dir}/installer-{u_id}-{history_version}", "w") as f: f.write(entity) else: u_id = installation.user_id or none installer_filepath = f"{team_installation_dir}/installer-{u_id}-latest" with open(installer_filepath, "w") as f: entity = json.dumps(installation.__dict__) f.write(entity) ``` -------------------------------- ### WebClient Initialization and Basic Usage Source: https://docs.slack.dev/tools/python-slack-sdk/reference/index.html Demonstrates how to initialize the WebClient with a token and make a simple chat.postMessage API call. ```python import os from slack_sdk import WebClient client = WebClient(token=os.environ['SLACK_API_TOKEN']) response = client.chat_postMessage( channel='#random', text="Hello world!") assert response["ok"] assert response["message"]["text"] == "Hello world!" ```