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)
• [x] 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}
        )