Add MongoDB installation store

[faisalsaifii]

I'd like to add MongoDB OAuth installation store which is currently not present in here

I recently developed my own app Pollify and wanted to use MongoDB for which there wasn't an installation store available, so I had to create one on my own.

Category

• slack_sdk.web.WebClient (sync/async) (Web API client)
• slack_sdk.webhook.WebhookClient (sync/async) (Incoming Webhook, response_url sender)
• slack_sdk.models (UI component builders)
• slack_sdk.oauth (OAuth Flow Utilities)
• slack_sdk.socket_mode (Socket Mode client)
• slack_sdk.audit_logs (Audit Logs API client)
• slack_sdk.scim (SCIM API client)
• slack_sdk.rtm (RTM client)
• slack_sdk.signature (Request Signature Verifier)

[seratch]

Hi @faisalsaifii, thank you so much for submitting this issue! Also, we are glad to know you've implemented your installation store with MongoDB.

Having more options out-of-the-box is great! However, our SDK maintenance team has a little bit limited capacity for this Python tool. Thefore, we'd like to hold off working on this addition until we receive more requests to the idea (i.e. a number of positive reactions from others). Please go ahead with your implementation for now.

[faisalsaifii]

Hi @seratch, sure I understand.

[MadhabaPatra]

@faisalsaifii Can you help me with the setup? I also wanted to implement mongodb installation store.

[faisalsaifii]

Hey @seratch, can I pick this up now?

[tingyiy]

this is my bot only implementation:

class MongoInstallationStore(InstallationStore):

def __init__(self):
    self.db = get_db()

def save(self, installation: Installation):
    self.db.slack_installations.update_one(
        {
            "enterprise_id": installation.enterprise_id,
            "team_id": installation.team_id,
        },
        {"$set": installation.to_dict()},
        upsert=True,
    )
    self.db.slack_bots.update_one(
        {"enterprise_id": installation.enterprise_id, "team_id": installation.team_id},
        {"$set": installation.to_bot().to_dict()},
        upsert=True,
    )

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:
        result = self.db.slack_installations.find_one(
            {"enterprise_id": enterprise_id, "user_id": user_id}
        )
        if result:
            del result["_id"]
            return Installation(**result)
        result = self.db.slack_installations.find_one({"enterprise_id": enterprise_id})
        if result:
            del result["_id"]
            return Installation(**result)
        return None
    result = self.db.slack_installations.find_one(
        {"team_id": team_id, "user_id": user_id}
    )
    if result:
        del result["_id"]
        return Installation(**result)
    result = self.db.slack_installations.find_one({"team_id": team_id})
    if result:
        del result["_id"]
        return Installation(**result)
    return None

def find_bot(
    self,
    *,
    enterprise_id: Optional[str],
    team_id: Optional[str],
    is_enterprise_install: Optional[bool] = False,
) -> Optional[Bot]:
    """Finds a bot scope installation per workspace / org"""
    if is_enterprise_install:
        result = self.db.slack_bots.find_one({"enterprise_id": enterprise_id})
        if result:
            del result["_id"]
            return Bot(**result)
        return None
    result = self.db.slack_bots.find_one({"team_id": team_id})
    if result:
        del result["_id"]
        return Bot(**result)
    return None

def delete_bot(
    self,
    *,
    enterprise_id: Optional[str],
    team_id: Optional[str],
) -> None:
    """Deletes a bot scope installation per workspace / org"""
    self.db.slack_bots.delete_one({"enterprise_id": enterprise_id, "team_id": team_id})

def delete_installation(
    self,
    *,
    enterprise_id: Optional[str],
    team_id: Optional[str],
    user_id: Optional[str] = None,
) -> None:
    """Deletes an installation that matches the given IDs"""
    if user_id is not None:
        self.db.slack_installations.delete_one(
            {"enterprise_id": enterprise_id, "team_id": team_id, "user_id": user_id}
        )
    else:
        self.db.slack_installations.delete_one(
            {"enterprise_id": enterprise_id, "team_id": team_id}
        )

[gustavo-bordin]

Here is my async approach in case anyone is interested

from pymongo.database import Database
from slack_sdk.oauth.installation_store.async_installation_store import AsyncInstallationStore
from slack_sdk.oauth.installation_store.models.bot import Bot
from slack_sdk.oauth.installation_store.models.installation import Installation



class MongoInstallationStore(AsyncInstallationStore):
    def __init__(self, db: Database[Dict[str, Any]]):
        self.db = db

    async def async_save(self, installation: Installation):
        self.db.slack_installations.update_one(
            {"enterprise_id": installation.enterprise_id, "team_id": installation.team_id},
            {"$set": installation.to_dict()},
            upsert=True,
        )

    async def async_save_bot(self, bot: Bot):
        self.db.slack_bots.update_one(
            {"enterprise_id": bot.enterprise_id, "team_id": bot.team_id},
            {"$set": bot.to_dict()},
            upsert=True,
        )

    async def async_find_bot(
        self,
        *,
        enterprise_id: Optional[str],
        team_id: Optional[str],
        is_enterprise_install: Optional[bool] = False,
    ) -> Optional[Bot]:
        if is_enterprise_install:
            result = self.db.slack_bots.find_one({"enterprise_id": enterprise_id})
            if result:
                del result["_id"]
                return Bot(**result)
            return None
        
        result = self.db.slack_bots.find_one({"team_id": team_id})

        if result:
            del result["_id"]
            return Bot(**result)
        
        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]:
        if is_enterprise_install:
            result = self.db.slack_installations.find_one(
                {"enterprise_id": enterprise_id, "user_id": user_id}
            )
            if result:
                del result["_id"]
                return Installation(**result)
            
            result = self.db.slack_installations.find_one({"enterprise_id": enterprise_id})
        
            if result:
                del result["_id"]
                return Installation(**result)

            return None
        
        result = self.db.slack_installations.find_one(
            {"team_id": team_id, "user_id": user_id}
        )

        if result:
            del result["_id"]
            return Installation(**result)
        
        result = self.db.slack_installations.find_one({"team_id": team_id})

        if result:
            del result["_id"]
            return Installation(**result)
        
        return None

    async def async_delete_bot(
        self,
        *,
        enterprise_id: Optional[str],
        team_id: Optional[str],
    ) -> None:
        self.db.slack_bots.delete_one({"enterprise_id": enterprise_id, "team_id": team_id})

    async def async_delete_installation(
        self,
        *,
        enterprise_id: Optional[str],
        team_id: Optional[str],
        user_id: Optional[str] = None,
    ) -> None:
        if user_id is not None:
            self.db.slack_installations.delete_one(
                {"enterprise_id": enterprise_id, "team_id": team_id, "user_id": user_id}
            )
        else:
            self.db.slack_installations.delete_one(
                {"enterprise_id": enterprise_id, "team_id": team_id}
            )

    async def async_delete_all(
        self,
        *,
        enterprise_id: Optional[str],
        team_id: Optional[str],
    ):
        await self.async_delete_bot(enterprise_id=enterprise_id, team_id=team_id)
        await self.async_delete_installation(enterprise_id=enterprise_id, team_id=team_id)`