slack

admin

ConversationAdmin

Bases: ModelAdmin[Conversation]

get_queryset

get_queryset(request: HttpRequest) -> QuerySet[Conversation]

Restrict the queryset to only include conversations that are not IncidentChannels. Incident channels are managed in the IncidentChannelAdmin.

Source code in src/firefighter/slack/admin.py

def get_queryset(self, request: HttpRequest) -> QuerySet[Conversation]:
    """Restrict the queryset to only include conversations that are not IncidentChannels. Incident channels are managed in the IncidentChannelAdmin."""
    qs = super().get_queryset(request)

    return Conversation.objects.not_incident_channel(qs)

ask_key_timestamps

ask_key_timestamps(self: firefighter.incidents.admin.IncidentAdmin, request: HttpRequest, queryset: QuerySet[Incident]) -> None

Will send a message to the Incident conversation (if it exists) to ask for key events. TODO Error handling.

Source code in src/firefighter/slack/admin.py

@action(description=_("Send a message to ask for key timestamps"))
def ask_key_timestamps(
    self: firefighter.incidents.admin.IncidentAdmin,
    request: HttpRequest,
    queryset: QuerySet[Incident],
) -> None:
    """Will send a message to the Incident conversation (if it exists) to ask for key events.
    TODO Error handling.
    """
    # ruff: noqa: PLC0415
    from firefighter.slack.views.modals.key_event_message import SlackMessageKeyEvents

    success: list[tuple[int, bool]] = []
    errors: list[tuple[int, bool, Exception | str]] = []
    for incident in queryset:
        if incident.conversation:
            try:
                incident.conversation.send_message_and_save(
                    SlackMessageKeyEvents(incident=incident)
                )
                success.append((incident.id, True))
            except SlackApiError as e:
                errors.append((incident.id, False, e))
        else:
            errors.append((incident.id, False, "No conversation"))

    if len(success) > 0:
        success_str = ", ".join(f"#{key[0]}" for key in success)
        self.message_user(
            request,
            ngettext(
                f"Sent message metrics for %d incident ({success_str}).",  # noqa: INT001
                f"Sent messages metrics for %d incidents ({success_str}).",
                len(success),
            )
            % len(success),
            constants.SUCCESS,
        )
    if len(errors) > 0:
        self.message_user(
            request,
            format_html(
                "Error sending message: <br/>{}",
                "<br/>".join(f"#{key[0]}: {key[2]}" for key in errors),
            ),
            constants.ERROR,
        )

factories

SlackProvider

Bases: BaseProvider

Custom Faker provider for generating Slack IDs.

forms

sos_form

messages

base

SlackMessageStrategy

Bases: Enum

Define how should the message be posted.

• append: add the message to the channel, even if a message with the same type has already been posted.
• replace: replace the last message with the same type (push new one, delete old one)
• update: update in place the last message with the same type

SlackMessageSurface

SlackMessageSurface()

Bases: ABC

Base class for Slack messages, which are sent to a channel or a user.

This provides a common interface to send messages with text, blocks and metadata.

This is helpful to send messages but also to save them in DB.

Source code in src/firefighter/slack/messages/base.py

def __init__(self) -> None:
    if not self.id or self.id == "":
        logger.warning(f"No Slack message ID set for {self.__class__.__name__}.")
    # Check the ID is well-formed to be a Slack event_type: alphanumeric string, starting with a letter, containing underscore
    elif not VALID_ID_REGEX.match(self.id):
        logger.warning(
            f"Slack message ID {self.id} is not well formed for {self.__class__.__name__}."
        )

strategy class-attribute instance-attribute

strategy: SlackMessageStrategy = SlackMessageStrategy.APPEND

Alphanumeric ID, starting with a letter, that may contain underscores.

get_blocks

get_blocks() -> list[Block]

Returns the blocks of the message.

Returns:

• list[Block] –

  list[Block]: List of Slack Blocks for the message. Default is the text as a SectionBlock.

Source code in src/firefighter/slack/messages/base.py

def get_blocks(self) -> list[Block]:
    """Returns the blocks of the message.

    Returns:
        list[Block]: List of Slack Blocks for the message. Default is the text as a SectionBlock.
    """
    return [SectionBlock(text=self.get_text())]

get_metadata

get_metadata() -> Metadata

The value of event_type should be an alphanumeric string, and human-readable. The value of this field may appear in the UI to developers, so keep this in mind when choosing a value. Developers should make an effort to name with the pattern and See more https://api.slack.com/reference/metadata.

Returns:

• Metadata ( Metadata ) –

  Slack Metadata for the message.

Source code in src/firefighter/slack/messages/base.py

def get_metadata(self) -> Metadata:
    """The value of `event_type` should be an alphanumeric string, and human-readable. The value of this field may appear in the UI to developers, so keep this in mind when choosing a value. Developers should make an effort to name with the pattern <resource_name_singular> and <action_in_past_tense>
    See more https://api.slack.com/reference/metadata.

    Returns:
        Metadata: Slack Metadata for the message.
    """
    return Metadata(event_type=self.id, event_payload={"ff_type": self.id})

get_text abstractmethod

get_text() -> str

Returns the text of the message.

Source code in src/firefighter/slack/messages/base.py

@abstractmethod
def get_text(self) -> str:
    """Returns the text of the message."""
    raise NotImplementedError

post_message

post_message(conversation_id: str, client: WebClient = DefaultWebClient, **kwargs: Never) -> SlackResponse

Deprecated. Only use for Global Channel, until the global channel is set in DB.

Source code in src/firefighter/slack/messages/base.py

@slack_client
def post_message(
    self,
    conversation_id: str,
    client: WebClient = DefaultWebClient,
    **kwargs: Never,
) -> SlackResponse:
    """Deprecated. Only use for Global Channel, until the global channel is set in DB."""
    return client.chat_postMessage(
        channel=conversation_id,
        **self.get_slack_message_params(),
    )

slack_messages

SlackMessageIncidentDeclaredAnnouncementGeneral

SlackMessageIncidentDeclaredAnnouncementGeneral(incident: Incident)

Bases: SlackMessageSurface

Parameters:

• incident (Incident) –

  Your incident

Source code in src/firefighter/slack/messages/slack_messages.py

def __init__(self, incident: Incident) -> None:
    """The message to post in general incident channel (tag=tech_incidents) when an incident is opened.

    Args:
        incident (Incident): Your incident
    """
    self.incident = incident
    super().__init__()

strategy class-attribute instance-attribute

strategy: SlackMessageStrategy = SlackMessageStrategy.APPEND

Alphanumeric ID, starting with a letter, that may contain underscores.

get_metadata

get_metadata() -> Metadata

The value of event_type should be an alphanumeric string, and human-readable. The value of this field may appear in the UI to developers, so keep this in mind when choosing a value. Developers should make an effort to name with the pattern and See more https://api.slack.com/reference/metadata.

Returns:

• Metadata ( Metadata ) –

  Slack Metadata for the message.

Source code in src/firefighter/slack/messages/base.py

def get_metadata(self) -> Metadata:
    """The value of `event_type` should be an alphanumeric string, and human-readable. The value of this field may appear in the UI to developers, so keep this in mind when choosing a value. Developers should make an effort to name with the pattern <resource_name_singular> and <action_in_past_tense>
    See more https://api.slack.com/reference/metadata.

    Returns:
        Metadata: Slack Metadata for the message.
    """
    return Metadata(event_type=self.id, event_payload={"ff_type": self.id})

post_message

post_message(conversation_id: str, client: WebClient = DefaultWebClient, **kwargs: Never) -> SlackResponse

Deprecated. Only use for Global Channel, until the global channel is set in DB.

Source code in src/firefighter/slack/messages/base.py

@slack_client
def post_message(
    self,
    conversation_id: str,
    client: WebClient = DefaultWebClient,
    **kwargs: Never,
) -> SlackResponse:
    """Deprecated. Only use for Global Channel, until the global channel is set in DB."""
    return client.chat_postMessage(
        channel=conversation_id,
        **self.get_slack_message_params(),
    )

SlackMessageIncidentRolesUpdated

SlackMessageIncidentRolesUpdated(
    incident: Incident, incident_update: IncidentUpdate | None, *, first_update: bool = False, updated_fields: list[str] | None = None
)

Bases: SlackMessageSurface

The message to post in the incident channel when the roles are updated.

Parameters:

• incident (Incident) –

  Your incident.

• incident_update (IncidentUpdate) –

  The opening incident update.

• first_update (bool, default: False ) –

  Whether this is the first update of the incident. Defaults to False.

• updated_fields (list[str], default: None ) –

  The fields that were updated. Defaults to None.

Source code in src/firefighter/slack/messages/slack_messages.py

def __init__(
    self,
    incident: Incident,
    incident_update: IncidentUpdate | None,
    *,
    first_update: bool = False,
    updated_fields: list[str] | None = None,
) -> None:
    self.incident = incident
    self.incident_update = incident_update
    self.first_update = first_update
    self.updated_fields = updated_fields
    self._new_roles = self._get_updated_roles()
    super().__init__()

strategy class-attribute instance-attribute

strategy: SlackMessageStrategy = SlackMessageStrategy.APPEND

Alphanumeric ID, starting with a letter, that may contain underscores.

post_message

post_message(conversation_id: str, client: WebClient = DefaultWebClient, **kwargs: Never) -> SlackResponse

Deprecated. Only use for Global Channel, until the global channel is set in DB.

Source code in src/firefighter/slack/messages/base.py

@slack_client
def post_message(
    self,
    conversation_id: str,
    client: WebClient = DefaultWebClient,
    **kwargs: Never,
) -> SlackResponse:
    """Deprecated. Only use for Global Channel, until the global channel is set in DB."""
    return client.chat_postMessage(
        channel=conversation_id,
        **self.get_slack_message_params(),
    )

models

conversation

Conversation

Bases: Model

Model a Slack API Conversation. A Slack Conversation can be a public channel, a private channel, a direct message, or a multi-person direct message. Reference: https://api.slack.com/types/conversation.

deep_link property

deep_link: str

Deep link (slack://) to the conversation in the Slack client.

link property

link: str

Regular HTTPS link to the conversation through Slack.com.

add_bookmark

add_bookmark(
    title: str,
    _type: str = "link",
    emoji: str | None = None,
    entity_id: str | None = None,
    link: str | None = None,
    parent_id: str | None = None,
    client: WebClient = DefaultWebClient,
    **kwargs: Any
) -> None

Convenience method to add a bookmark on this conversation.

Source code in src/firefighter/slack/models/conversation.py

@slack_client
def add_bookmark(
    self,
    title: str,
    _type: str = "link",
    emoji: str | None = None,
    entity_id: str | None = None,
    link: str | None = None,  # include when type is 'link'
    parent_id: str | None = None,
    client: WebClient = DefaultWebClient,
    **kwargs: Any,
) -> None:
    """Convenience method to add a bookmark on this conversation."""
    client.bookmarks_add(
        channel_id=self.channel_id,
        title=title,
        link=link,
        emoji=emoji,
        type=_type,
        entity_id=entity_id,
        parent_id=parent_id,
        **kwargs,
    )

send_message

send_message(
    text: str | None = None, blocks: Sequence[dict[str, Any] | Block] | None = None, client: WebClient = DefaultWebClient, **kwargs: Any
) -> None

Convenience method to send a message on this conversation.

Source code in src/firefighter/slack/models/conversation.py

@slack_client
def send_message(
    self,
    text: str | None = None,
    blocks: Sequence[dict[str, Any] | Block] | None = None,
    client: WebClient = DefaultWebClient,
    **kwargs: Any,
) -> None:
    """Convenience method to send a message on this conversation."""
    if kwargs.get("channel"):
        raise ValueError(
            "You can't set the channel when using send_message on a conversation!"
        )

    client.chat_postMessage(
        channel=self.channel_id, text=text, blocks=blocks, **kwargs
    )

send_message_and_save

send_message_and_save(
    message: SlackMessageSurface,
    client: WebClient = DefaultWebClient,
    strategy: SlackMessageStrategy | None = None,
    strategy_args: dict[str, Any] | None = None,
    *,
    pin: bool = False
) -> SlackResponse

Convenience method to send a message on this conversation.

Source code in src/firefighter/slack/models/conversation.py

@slack_client
def send_message_and_save(
    self,
    message: SlackMessageSurface,
    client: WebClient = DefaultWebClient,
    strategy: SlackMessageStrategy | None = None,
    strategy_args: dict[str, Any] | None = None,
    *,
    pin: bool = False,
) -> SlackResponse:
    """Convenience method to send a message on this conversation."""
    if strategy is None:
        strategy = message.strategy

    kwargs = {}
    # XXX Get save context?
    if hasattr(message, "incident"):
        kwargs["incident"] = message.incident
    if hasattr(message, "incident_update"):
        kwargs["incident_update"] = message.incident_update
    kwargs["ff_type"] = message.id
    if strategy == SlackMessageStrategy.APPEND:
        strategy_res = self._send_message_strategy_append(message, client, kwargs)
    elif strategy == SlackMessageStrategy.UPDATE:
        strategy_res = self._send_message_strategy_update(message, client, kwargs)
    elif strategy == SlackMessageStrategy.REPLACE:
        strategy_res = self._send_message_strategy_replace(
            message, client, strategy_args, kwargs
        )
    if pin:
        self._pin_message(
            res=strategy_res,
            client=client,
        )
    return strategy_res

send_message_ephemeral

send_message_ephemeral(message: SlackMessageSurface, user: SlackUser | str, client: WebClient = DefaultWebClient, **kwargs: Any) -> None

Convenience method to send an ephemeral message on this conversation. user, channel, blocks, text and metadata should not be passed in kwargs.

Source code in src/firefighter/slack/models/conversation.py

@slack_client
def send_message_ephemeral(
    self,
    message: SlackMessageSurface,
    user: SlackUser | str,
    client: WebClient = DefaultWebClient,
    **kwargs: Any,
) -> None:
    """Convenience method to send an ephemeral message on this conversation.
    `user`, `channel`, `blocks`, `text` and `metadata` should not be passed in kwargs.
    """
    user_id = user.slack_id if isinstance(user, SlackUser) else user

    client.chat_postEphemeral(
        user=user_id,
        channel=self.channel_id,
        **(kwargs | message.get_slack_message_params()),
    )

incident_channel

IncidentChannel

Bases: Conversation

deep_link property

deep_link: str

Deep link (slack://) to the conversation in the Slack client.

link property

link: str

Regular HTTPS link to the conversation through Slack.com.

add_bookmark

add_bookmark(
    title: str,
    _type: str = "link",
    emoji: str | None = None,
    entity_id: str | None = None,
    link: str | None = None,
    parent_id: str | None = None,
    client: WebClient = DefaultWebClient,
    **kwargs: Any
) -> None

Convenience method to add a bookmark on this conversation.

Source code in src/firefighter/slack/models/conversation.py

@slack_client
def add_bookmark(
    self,
    title: str,
    _type: str = "link",
    emoji: str | None = None,
    entity_id: str | None = None,
    link: str | None = None,  # include when type is 'link'
    parent_id: str | None = None,
    client: WebClient = DefaultWebClient,
    **kwargs: Any,
) -> None:
    """Convenience method to add a bookmark on this conversation."""
    client.bookmarks_add(
        channel_id=self.channel_id,
        title=title,
        link=link,
        emoji=emoji,
        type=_type,
        entity_id=entity_id,
        parent_id=parent_id,
        **kwargs,
    )

invite_users

invite_users(users_mapped: list[User], client: WebClient = DefaultWebClient) -> None

Invite users to the conversation, if they have a Slack user linked and are active.

Try to invite all users as batch, but if some fail, continue individually.

Source code in src/firefighter/slack/models/incident_channel.py

@slack_client
def invite_users(
    self, users_mapped: list[User], client: WebClient = DefaultWebClient
) -> None:
    """Invite users to the conversation, if they have a Slack user linked and are active.

    Try to invite all users as batch, but if some fail, continue individually.
    """
    users_with_slack: list[User] = self._get_active_slack_users(users_mapped)
    users_with_slack = list(set(users_with_slack))  # Remove duplicates

    user_id_list: set[str] = self._get_slack_id_list(users_with_slack)
    if not user_id_list:
        logger.info(f"No users to invite to the conversation {self}.")
        return

    logger.info(f"Inviting users with SlackIDs: {user_id_list}")

    invited_slack_user_ids = self._invite_users_to_conversation(
        user_id_list, client
    )
    invited_users = {
        u
        for u in users_with_slack
        if u.slack_user and u.slack_user.slack_id in invited_slack_user_ids
    }

    if invited_slack_user_ids != set(user_id_list):
        logger.warning(
            f"Could not invite all users to the conversation {self}. Missing users: {user_id_list - invited_slack_user_ids }"
        )

    self.incident.members.add(*invited_users)

send_message

send_message(
    text: str | None = None, blocks: Sequence[dict[str, Any] | Block] | None = None, client: WebClient = DefaultWebClient, **kwargs: Any
) -> None

Convenience method to send a message on this conversation.

Source code in src/firefighter/slack/models/conversation.py

@slack_client
def send_message(
    self,
    text: str | None = None,
    blocks: Sequence[dict[str, Any] | Block] | None = None,
    client: WebClient = DefaultWebClient,
    **kwargs: Any,
) -> None:
    """Convenience method to send a message on this conversation."""
    if kwargs.get("channel"):
        raise ValueError(
            "You can't set the channel when using send_message on a conversation!"
        )

    client.chat_postMessage(
        channel=self.channel_id, text=text, blocks=blocks, **kwargs
    )

send_message_and_save

send_message_and_save(
    message: SlackMessageSurface,
    client: WebClient = DefaultWebClient,
    strategy: SlackMessageStrategy | None = None,
    strategy_args: dict[str, Any] | None = None,
    *,
    pin: bool = False
) -> SlackResponse

Convenience method to send a message on this conversation.

Source code in src/firefighter/slack/models/conversation.py

@slack_client
def send_message_and_save(
    self,
    message: SlackMessageSurface,
    client: WebClient = DefaultWebClient,
    strategy: SlackMessageStrategy | None = None,
    strategy_args: dict[str, Any] | None = None,
    *,
    pin: bool = False,
) -> SlackResponse:
    """Convenience method to send a message on this conversation."""
    if strategy is None:
        strategy = message.strategy

    kwargs = {}
    # XXX Get save context?
    if hasattr(message, "incident"):
        kwargs["incident"] = message.incident
    if hasattr(message, "incident_update"):
        kwargs["incident_update"] = message.incident_update
    kwargs["ff_type"] = message.id
    if strategy == SlackMessageStrategy.APPEND:
        strategy_res = self._send_message_strategy_append(message, client, kwargs)
    elif strategy == SlackMessageStrategy.UPDATE:
        strategy_res = self._send_message_strategy_update(message, client, kwargs)
    elif strategy == SlackMessageStrategy.REPLACE:
        strategy_res = self._send_message_strategy_replace(
            message, client, strategy_args, kwargs
        )
    if pin:
        self._pin_message(
            res=strategy_res,
            client=client,
        )
    return strategy_res

send_message_ephemeral

send_message_ephemeral(message: SlackMessageSurface, user: SlackUser | str, client: WebClient = DefaultWebClient, **kwargs: Any) -> None

Convenience method to send an ephemeral message on this conversation. user, channel, blocks, text and metadata should not be passed in kwargs.

Source code in src/firefighter/slack/models/conversation.py

@slack_client
def send_message_ephemeral(
    self,
    message: SlackMessageSurface,
    user: SlackUser | str,
    client: WebClient = DefaultWebClient,
    **kwargs: Any,
) -> None:
    """Convenience method to send an ephemeral message on this conversation.
    `user`, `channel`, `blocks`, `text` and `metadata` should not be passed in kwargs.
    """
    user_id = user.slack_id if isinstance(user, SlackUser) else user

    client.chat_postEphemeral(
        user=user_id,
        channel=self.channel_id,
        **(kwargs | message.get_slack_message_params()),
    )

message

Message

Bases: Model

Model a Slack API Conversation. A Slack Conversation can be a public channel, a private channel, a direct message, or a multi-person direct message. Reference: https://api.slack.com/types/conversation.

sos

Sos

Bases: Model

A SOS is a target with a name, a conversation, and optionally a user group.

Incident responders can use it with /incident sos to ask for help to a specific group of people.

Helpers will be notified in the selected conversation, and the user group will be mentioned (or @here if no user group is selected)

usergroup_slack_fmt property

usergroup_slack_fmt: str

Returns either @usergroup or @here depending on usergroup presence.

user

SlackUser

Bases: Model

Holds data about a Slack User, linked to an :model:incidents.user. slack_id field is not used as PK, as it is only guaranteed by Slack to be unique in pair with a team_id.

url property

url: str

Returns an HTTPS ULR to the Slack user's profile. For deep linking, use link instead.

send_private_message

send_private_message(message: SlackMessageSurface, client: WebClient = DefaultWebClient, **kwargs: Any) -> None

Send a private message to the user.

Source code in src/firefighter/slack/models/user.py

@slack_client
def send_private_message(
    self,
    message: SlackMessageSurface,
    client: WebClient = DefaultWebClient,
    **kwargs: Any,
) -> None:
    """Send a private message to the user."""
    client.chat_postMessage(
        channel=self.slack_id,
        text=message.get_text(),
        metadata=message.get_metadata(),
        blocks=message.get_blocks(),
        **kwargs,
    )

SlackUserManager

Bases: Manager['SlackUser']

get_user_by_slack_id

get_user_by_slack_id(slack_id: str, defaults: dict[str, str] | None = None, client: WebClient = DefaultWebClient) -> User | None

Returns a User from DB if it exists, or fetch its info from Slack, save it to DB and returns it.

Source code in src/firefighter/slack/models/user.py

@slack_client
def get_user_by_slack_id(
    self,
    slack_id: str,
    defaults: dict[str, str] | None = None,
    client: WebClient = DefaultWebClient,
) -> User | None:
    """Returns a User from DB if it exists, or fetch its info from Slack, save it to DB and returns it."""
    if not slack_id:
        raise ValueError("slack_id cannot be empty")

    # Try fetching it from DB...
    try:
        slack_user = self.select_related("user").get(slack_id=slack_id)
        return slack_user.user  # noqa: TRY300
    except SlackUser.DoesNotExist:
        slack_user = None

    if (
        defaults
        and "email" in defaults
        and "name" in defaults
        and defaults["name"]
        and defaults["email"]
    ):
        user, _ = User.objects.get_or_create(
            username=defaults["email"].split("@")[0],
            email=defaults["email"],
            defaults={"name": defaults["name"]},
        )
        slack_user, _created = SlackUser.objects.get_or_create(
            slack_id=slack_id, user=user, defaults={"id": uuid.uuid4()}
        )
        return user

    # If not in DB, fetch the user's info from firefighter.slack...
    logger.debug("Fetch user from Slack")
    try:
        user_info = client.users_info(user=slack_id)
        if not user_info.get("ok"):
            logger.error("Could not fetch user from firefighter.slack.")
            return None
    except slack_sdk.errors.SlackApiError:
        logger.exception(f"Could not find Slack user with ID: {slack_id}")
        return None

    clean_user_info = self.unpack_user_info(user_info=user_info)

    if "email" not in clean_user_info and "display_name" not in clean_user_info:
        logger.error(
            f"Not enough info in Slack user.info response! user_info: {user_info}, parsed: {clean_user_info}"
        )
        return None
    user, _ = User.objects.get_or_create(
        username=clean_user_info["email"].split("@")[0],
        email=clean_user_info["email"],
        defaults={"name": clean_user_info["name"]},
    )
    if user is None:
        raise ValueError("Could not create user from Slack info")

    # TODO Handle case of user but with outdated email
    logger.debug("Creating Slack user")
    logger.debug(user_info)
    slack_user, _ = self.get_or_create_from_slack(user_info=user_info, user=user)
    if not slack_user:
        logger.error("Could not upsert slack_user in DB WTF")
        return None

    return user

unpack_user_info staticmethod

unpack_user_info(user_info: SlackResponse | dict[str, Any]) -> dict[str, Any]

Returns a dict contains fields for the SlackUser, from a SlackResponse. email, name and id should always be returned.

Source code in src/firefighter/slack/models/user.py

@staticmethod
def unpack_user_info(user_info: SlackResponse | dict[str, Any]) -> dict[str, Any]:
    """Returns a dict contains fields for the SlackUser, from a SlackResponse.
    email, name and id should always be returned.
    """
    user_args = {}
    if user_info.get("user") and user_info["user"] is not None:
        user_info = cast("dict[str, Any]", user_info["user"])

    user_args["slack_id"] = get_in(user_info, "id")
    user_args["first_name"] = get_in(user_info, "profile.first_name")
    user_args["last_name"] = get_in(user_info, "profile.last_name")
    user_args["deleted"] = get_in(user_info, "deleted")

    # TODO Fix mess with name vs username
    username = get_in(user_info, "name")
    if username:
        user_args["username"] = username

    if not get_in(user_info, "is_bot") and get_in(user_info, "id") != "USLACKBOT":
        user_args["email"] = get_in(user_info, "profile.email")
        user_args["name"] = get_in(user_info, "profile.real_name")
    else:
        user_args["email"] = get_in(user_info, "name")
        user_args["name"] = get_in(user_info, "profile.real_name")

    if get_in(user_info, "profile.image_512"):
        avatar = get_in(user_info, "profile.image_512")
        if avatar and len(
            avatar
        ):  # <= SlackUser._meta.get_field('image').max_length:
            user_args["image"] = avatar
    elif get_in(user_info, "profile.image_192"):
        avatar = get_in(user_info, "profile.image_192")
        if avatar and len(
            avatar
        ):  # <= SlackUser._meta.get_field('image').max_length:
            user_args["image"] = avatar

    if user_args["first_name"] is None:
        user_args["first_name"] = user_args["name"].split(" ")[0]
    if user_args["last_name"] is None:
        user_args["last_name"] = user_args["name"].split(" ")[-1]
    return user_args

upsert_by_email

upsert_by_email(email: str, client: WebClient = DefaultWebClient) -> User | None

Returns a User from DB if it exists, or fetch its info from Slack, save it to DB and returns it.

Source code in src/firefighter/slack/models/user.py

@slack_client
def upsert_by_email(
    self,
    email: str,
    client: WebClient = DefaultWebClient,
) -> User | None:
    """Returns a User from DB if it exists, or fetch its info from Slack, save it to DB and returns it."""
    logger.debug(f"Looking for user by email: {email}")
    # Try fetching it from DB...
    try:
        user = User.objects.get(email=email)
    except User.DoesNotExist:
        logger.info("No user in DB. Fetching it from Slack...")
    else:
        return user

    # If not in DB, fetch the user's info from firefighter.slack...
    try:
        user_info = client.users_lookupByEmail(email=email)
        if not user_info.get("ok"):
            logger.error(f"Could not fetch user from Slack. User: email={email}")
            return None
    except slack_sdk.errors.SlackApiError:
        logger.exception(f"Could not find Slack user with email: {email}")
        return None

    logger.debug(user_info)

    clean_user_info = self.unpack_user_info(user_info=user_info)

    slack_user = SlackUser.objects.get_or_none(slack_id=clean_user_info["slack_id"])

    # If we have a SlackUser but not user with email => Update user email
    if slack_user:
        user = slack_user.user
        user.email = email
        user.save()
        return user

    # If we have no Slack User, let's go ahead and create a User and its associated SlackUser
    user, _created = User.objects.get_or_create(
        email=email,
        username=email.split("@")[0],
        defaults={
            "name": clean_user_info["name"],
        },
    )

    try:
        slack_user, _created = self.get_or_create_from_slack(
            user_info=user_info,
            slack_id=clean_user_info["slack_id"],
            defaults={"user_id": user.id, "id": uuid.uuid4()},
        )
        if slack_user and slack_user.user.email == email:
            return user
        logger.warning(f"1. Change of mail for user: {clean_user_info['slack_id']}")

    except IntegrityError:
        logger.warning(
            f"2. Change of mail for user: {clean_user_info['slack_id']}",
            exc_info=True,
        )
        slack_user, _created = self.get_or_create_from_slack(
            user_info=user_info,
            user_id=user.id,
            defaults={"slack_id": clean_user_info["slack_id"], "id": uuid.uuid4()},
        )
        return user
    return None

user_group

UserGroup

Bases: Model

Model a Slack API UserGroup. Reference: https://api.slack.com/types/usergroup.

link property

link: str

Regular HTTPS link to the conversation through Slack.com.

UserGroupManager

Bases: Manager['UserGroup']

fetch_all_usergroups_data staticmethod

fetch_all_usergroups_data(client: WebClient = DefaultWebClient, *, include_users: bool = False) -> list[dict[str, Any]]

Fetch all usergroups from firefighter.slack.

Returns the list of usergroups

Source code in src/firefighter/slack/models/user_group.py

@staticmethod
def fetch_all_usergroups_data(
    client: WebClient = DefaultWebClient,
    *,
    include_users: bool = False,
) -> list[dict[str, Any]]:
    """Fetch all usergroups from firefighter.slack.

    Returns the list of usergroups
    """
    slack_response_usergroups = client.usergroups_list(include_users=include_users)

    ug_list = get_in(slack_response_usergroups, "usergroups")
    if not isinstance(ug_list, list):
        err_msg = (
            f"Expected usergroups to be a list, but got {type(ug_list)}: {ug_list}"
        )

        raise TypeError(err_msg)
    return ug_list

fetch_usergroup staticmethod

fetch_usergroup(
    group_slack_id: str | None = None, group_handle: str | None = None, client: WebClient = DefaultWebClient, **kwargs: Any
) -> UserGroup | None

Import a "usergroup" from Slack and return its UserGroup model. Either group_slack_id or group_handle must be provided.

Parameters:

• group_slack_id (str | None, default: None ) –

  The Slack usergroup id of the usergroup to import. Usually starts with S. Defaults to None.

• group_handle (str | None, default: None ) –

  Handle (@xxx) of the usergroup to import. Defaults to None.

• client (WebClient, default: DefaultWebClient ) –

  Slack client. Defaults to DefaultWebClient.

• **kwargs (Any, default: {} ) –

  Additional keyword arguments to pass to created UserGroup model.

Returns:

• UserGroup | None –

  UserGroup | None: UserGroup model or None if not found

Source code in src/firefighter/slack/models/user_group.py

@staticmethod
@slack_client
def fetch_usergroup(
    group_slack_id: str | None = None,
    group_handle: str | None = None,
    client: WebClient = DefaultWebClient,
    **kwargs: Any,
) -> UserGroup | None:
    """Import a "usergroup" from Slack and return its UserGroup model.
    Either group_slack_id or group_handle must be provided.

    Args:
        group_slack_id (str | None, optional): The Slack usergroup id of the usergroup to import. Usually starts with `S`. Defaults to None.
        group_handle (str | None, optional): Handle (@xxx) of the usergroup to import. Defaults to None.
        client (WebClient, optional): Slack client. Defaults to DefaultWebClient.
        **kwargs: Additional keyword arguments to pass to created UserGroup model.

    Returns:
        UserGroup | None: UserGroup model or None if not found
    """
    slack_response_usergroups = UserGroupManager.fetch_all_usergroups_data(
        client=client
    )

    usergroup = UserGroupManager.get_usergroup_data_from_list(
        usergroups=slack_response_usergroups,
        group_slack_id=group_slack_id,
        group_handle=group_handle,
    )
    if not usergroup:
        logger.warning(
            f"Could not find matching group! group_slack_id: {group_slack_id}; group_handle: {group_handle}"
        )
        return None
    logger.debug(usergroup)

    return UserGroup(
        **UserGroupManager.parse_slack_response(usergroup),
        **kwargs,
    )

rules

Rules for publishing incidents in channels.

This module may be removed in a future version.

signals

create_incident_conversation

This module contains the logic to open an incident channel and invite responders.

XXX It might be divided into two signals/tasks, one for creating, one for inviting. XXX Sending the end signal should be done by this signal's caller, not by this signal receiver directly.

create_incident_slack_conversation

create_incident_slack_conversation(incident: Incident, *_args: Any, **_kwargs: Any) -> None | int

Main process to open an incident channel, set it up and invite responders. It MUST be called when an incident is created.

Parameters:

• incident (Incident) –

  The incident to open. It should be saved before calling this function, and have its first incident update created.

Source code in src/firefighter/slack/signals/create_incident_conversation.py

@receiver(signal=create_incident_conversation)
def create_incident_slack_conversation(
    incident: Incident,
    *_args: Any,
    **_kwargs: Any,
) -> None | int:
    """Main process to open an incident channel, set it up and invite responders. It MUST be called when an incident is created.

    Args:
        incident (Incident): The incident to open. It should be saved before calling this function, and have its first incident update created.

    """
    channel: IncidentChannel | None = IncidentChannel.objects.create_incident_channel(
        incident=incident
    )
    if not channel:
        logger.warning("Giving up on Slack channel creation for %s.", incident.id)
        return None

    new_channel_id: str = channel.channel_id
    if not new_channel_id:
        logger.warning("Missing channel id for %s.", incident.id)
        return None
    if not (
        incident.created_by
        and hasattr(incident.created_by, "slack_user")
        and incident.created_by.slack_user
    ):
        if not hasattr(incident.created_by, "slack_user"):
            SlackUser.objects.add_slack_id_to_user(incident.created_by)
        logger.debug(incident.created_by)

    # Join the channel. We are already in the channel if it is a private channel.
    if not incident.private:
        channel.conversations_join()

    channel.set_incident_channel_topic()

    # Add the person that opened the incident in the channel
    if (
        incident.created_by
        and hasattr(incident.created_by, "slack_user")
        and incident.created_by.slack_user
        and incident.created_by.slack_user.slack_id
    ):
        try:
            channel.invite_users([incident.created_by])
        except SlackApiError:
            logger.warning(
                f"Could not import Slack opener user! Slack ID: {incident.created_by.slack_user.slack_id}, User {incident.created_by}, Channel ID {new_channel_id}",
                exc_info=True,
            )
    else:
        logger.warning("Could not find user Slack ID for opener_user!")

    # Send message in the created channel
    channel.send_message_and_save(
        SlackMessageIncidentDeclaredAnnouncement(incident), pin=True
    )

    # Post in general channel #tech-incidents if needed
    if should_publish_in_general_channel(incident, incident_update=None):
        announcement_general = SlackMessageIncidentDeclaredAnnouncementGeneral(incident)

        tech_incidents_conversation = Conversation.objects.get_or_none(
            tag="tech_incidents"
        )
        if tech_incidents_conversation:
            tech_incidents_conversation.send_message_and_save(announcement_general)
        else:
            logger.warning(
                "Could not find tech_incidents conversation! Is there a channel with tag tech_incidents?"
            )

    # Create a response team
    users_list: list[User] = incident.build_invite_list()

    # Invite all users
    incident.conversation.invite_users(users_list)

    # Post in #it-deploy if needed
    if should_publish_in_it_deploy_channel(incident):
        announcement_it_deploy = SlackMessageDeployWarning(incident)
        announcement_it_deploy.id = f"{announcement_it_deploy.id}_{incident.id}"

        it_deploy_conversation = Conversation.objects.get_or_none(tag="it_deploy")
        if it_deploy_conversation:
            it_deploy_conversation.send_message_and_save(announcement_it_deploy)
        else:
            logger.warning(
                "Could not find it_deploy conversation! Is there a channel with tag it_deploy?"
            )

    incident_channel_done.send_robust(
        sender=__name__,
        incident=incident,
        channel=channel,
    )
    return None

get_users

get_invites_from_slack

get_invites_from_slack(incident: Incident, **_kwargs: Any) -> list[User]

New version using cached users instead of querying Slack API.

Source code in src/firefighter/slack/signals/get_users.py

@receiver(signal=signals.get_invites)
def get_invites_from_slack(incident: Incident, **_kwargs: Any) -> list[User]:
    """New version using cached users instead of querying Slack API."""
    # Prepare sub-queries
    slack_usergroups: QuerySet[UserGroup] = incident.component.usergroups.all()
    slack_conversations: QuerySet[Conversation] = incident.component.conversations.all()

    # We make sure to exclude the bot user, and avoid duplicates with distinct()
    # Also make sure that all users have related SlackUser and a slack_id
    queryset = (
        User.objects.filter(slack_user__isnull=False)
        .exclude(slack_user__slack_id=SlackApp().details["user_id"])
        .exclude(slack_user__slack_id="")
        .exclude(slack_user__slack_id__isnull=True)
        .filter(
            Q(conversation__in=slack_conversations) | Q(usergroup__in=slack_usergroups)
        )
        .distinct()
    )
    return list(queryset)

handle_incident_channel_done

incident_closed

incident_updated

publish_status_update

publish_status_update(
    incident: Incident, incident_update: IncidentUpdate, *, status_changed: bool = False, old_priority: Priority | None = None
) -> None

Publishes an update to the incident status.

Source code in src/firefighter/slack/signals/incident_updated.py

def publish_status_update(
    incident: Incident,
    incident_update: IncidentUpdate,
    *,
    status_changed: bool = False,
    old_priority: Priority | None = None,
) -> None:
    """Publishes an update to the incident status."""
    message = SlackMessageIncidentStatusUpdated(
        incident=incident,
        incident_update=incident_update,
        in_channel=True,
    )
    incident.conversation.send_message_and_save(message)

    # Post to #tech-incidents
    if should_publish_in_general_channel(
        incident=incident, incident_update=incident_update, old_priority=old_priority
    ):
        publish_update_in_general_channel(
            incident=incident,
            incident_update=incident_update,
            status_changed=status_changed,
            old_priority=old_priority,
        )

    if (
        incident.ask_for_milestones
        and status_changed
        and incident.status >= IncidentStatus.FIXED
    ):
        from firefighter.slack.views.modals.key_event_message import (  # noqa: PLC0415
            SlackMessageKeyEvents,
        )

        incident.conversation.send_message_and_save(
            SlackMessageKeyEvents(incident=incident)
        )

    if should_publish_in_it_deploy_channel(incident=incident):
        announcement_it_deploy = SlackMessageDeployWarning(incident)
        announcement_it_deploy.id = f"{announcement_it_deploy.id}_{incident.id}"

        it_deploy_conversation = Conversation.objects.get_or_none(tag="it_deploy")
        if it_deploy_conversation:
            it_deploy_conversation.send_message_and_save(announcement_it_deploy)
        else:
            logger.warning(
                "Could not find it_deploy conversation! Is there a channel with tag it_deploy?"
            )

postmortem_created

roles_reminders

slack_app

SlackApp

Bases: App

Subclass of the Slack App, as a singleton.

slack_client

slack_client(function: Callable[P, R]) -> Callable[P, R]

Adds a Slack client in client kwargs, if none is provided.

Can be used as a decorator with @slack_client or as a function with slack_client(function).

Source code in src/firefighter/slack/slack_app.py

def slack_client(function: Callable[P, R]) -> Callable[P, R]:
    """Adds a Slack client in `client` kwargs, if none is provided.

    Can be used as a decorator with `@slack_client` or as a function with `slack_client(function)`.
    """

    def wrap_function(*args: P.args, **kwargs: P.kwargs) -> R:
        if "client" not in kwargs:
            kwargs["client"] = SlackApp().client
        return function(*args, **kwargs)

    return wrap_function

slack_incident_context

get_incident_from_app_home_element

get_incident_from_app_home_element(body: dict[str, Any]) -> Incident | None

Get an incident from an action, found in app home accessory.

Source code in src/firefighter/slack/slack_incident_context.py

@incident_resolve_strategy
def get_incident_from_app_home_element(body: dict[str, Any]) -> Incident | None:
    """Get an incident from an action, found in app home accessory."""
    action = get_first_in(
        body.get("actions", []), "selected_option.value", ("update_status", "open_link")
    )
    if action is None:
        return None
    incident_id = action.get("block_id").strip(  # noqa: B005
        "app_home_incident_element_"
    )

    try:
        incident_id = int(incident_id)
    except ValueError:
        logger.warning(
            f"Select incident ('{incident_id}') was not a valid incident ID!"
        )
        return None
    return Incident.objects.get(id=incident_id)

get_incident_from_body_channel_id_in_command

get_incident_from_body_channel_id_in_command(body: dict[str, Any]) -> Incident | None

Get an incident from a channel_id, found in commands.

Source code in src/firefighter/slack/slack_incident_context.py

@incident_resolve_strategy
def get_incident_from_body_channel_id_in_command(
    body: dict[str, Any]
) -> Incident | None:
    """Get an incident from a channel_id, found in commands."""
    channel_id = body.get("channel_id")
    if channel_id:
        channel = IncidentChannel.objects.filter(channel_id=channel_id).first()
        if channel and channel.incident:
            return channel.incident
    return None

get_incident_from_body_channel_id_in_message_shortcut

get_incident_from_body_channel_id_in_message_shortcut(body: dict[str, Any]) -> Incident | None

Get an incident from a channel.id, found in message shortcut.

Source code in src/firefighter/slack/slack_incident_context.py

@incident_resolve_strategy
def get_incident_from_body_channel_id_in_message_shortcut(
    body: dict[str, Any]
) -> Incident | None:
    """Get an incident from a channel.id, found in message shortcut."""
    channel_id = get_in(body, "channel.id")
    if channel_id:
        channel = IncidentChannel.objects.filter(channel_id=channel_id).first()
        if channel and channel.incident:
            return channel.incident
    return None

get_incident_from_button_value

get_incident_from_button_value(body: dict[str, Any]) -> Incident | Literal[False] | None

Try to get the incident ID from button value.

Source code in src/firefighter/slack/slack_incident_context.py

@incident_resolve_strategy
def get_incident_from_button_value(
    body: dict[str, Any]
) -> Incident | Literal[False] | None:
    """Try to get the incident ID from button value."""
    actions = body.get("actions")
    if not isinstance(actions, list):
        return False
    # XXX(dugab): this should not use an allow-list, but loop over values or use a specific prefix/suffix/pattern
    action = get_first_in(
        actions,
        key="action_id",
        matches={
            "open_modal_incident_update_roles",
            "open_modal_incident_update_status",
            "open_modal_downgrade_workflow",
        },
    )

    if not action:
        return False
    value = action.get("value")
    if not value:
        return False
    try:
        incident_id = int(value)
    except (ValueError, TypeError):
        logger.warning("Response.actions.action_id.value was not a valid incident ID!")
        return None
    return Incident.objects.get(id=incident_id)

get_incident_from_context

get_incident_from_context(body: dict[str, Any]) -> Incident | None

Returns an Incident or None, from a Slack body.

Source code in src/firefighter/slack/slack_incident_context.py

def get_incident_from_context(body: dict[str, Any]) -> Incident | None:
    """Returns an Incident or None, from a Slack body."""
    for strat in INCIDENT_RESOLVING_STRATEGIES:
        incident = strat(body)
        if incident:
            logger.debug(f"Found incident {incident.id} using {strat.__name__}")
            return incident

    logger.info(f"Incident could not be inferred from Slack body context: {body}")
    return None

get_incident_from_view_action

get_incident_from_view_action(body: dict[str, Any]) -> Literal[False] | None | Incident

Get incident id from select incident in modal.

Source code in src/firefighter/slack/slack_incident_context.py

@incident_resolve_strategy
def get_incident_from_view_action(
    body: dict[str, Any]
) -> Literal[False] | None | Incident:
    """Get incident id from select incident in modal."""
    view_type = body.get("type")

    if view_type not in {"view_submission", "block_actions"}:
        return False
    actions = body.get("actions")
    if not isinstance(actions, list):
        return False
    action = get_first_in(
        actions, key="action_id", matches={"incident_update_select_incident"}
    )

    if not action:
        return False

    select_incident_value = get_in(action, "selected_option.value")
    logger.debug(action)
    logger.debug(select_incident_value)

    if not select_incident_value:
        return False
    try:
        incident_id = int(select_incident_value)
    except ValueError:
        logger.warning(
            f"Select incident ('{select_incident_value}') was not a valid incident ID!"
        )
        return None
    return Incident.objects.get(id=incident_id)

get_incident_from_view_submission_metadata

get_incident_from_view_submission_metadata(body: dict[str, Any]) -> Incident | Literal[False] | None

Get incident id from view modal metadata.

Source code in src/firefighter/slack/slack_incident_context.py

@incident_resolve_strategy
def get_incident_from_view_submission_metadata(
    body: dict[str, Any]
) -> Incident | Literal[False] | None:
    """Get incident id from view modal metadata."""
    view_type = body.get("type")
    private_metadata = get_in(body, ["view", "private_metadata"])

    if not private_metadata:
        return False

    if view_type not in {"view_submission", "block_actions"}:
        return False

    try:
        incident_id = int(private_metadata)
    except ValueError:
        logger.warning(
            f"Response.view.private_metadata ('{private_metadata}') was not a valid incident ID!"
        )
        return None
    return Incident.objects.get(id=incident_id)

get_incident_from_view_submission_selected

get_incident_from_view_submission_selected(body: dict[str, Any]) -> Incident | Literal[False] | None

Get incident id from select incident in modal.

Source code in src/firefighter/slack/slack_incident_context.py

@incident_resolve_strategy
def get_incident_from_view_submission_selected(
    body: dict[str, Any]
) -> Incident | Literal[False] | None:
    """Get incident id from select incident in modal."""
    view_type = body.get("type")

    if view_type not in {"view_submission", "block_actions"}:
        return False

    select_incident_value = get_in(
        body,
        "view.state.values.incident_update_select_incident.incident_update_select_incident.selected_option.value",
    )
    if not select_incident_value:
        # Check the default value (appears selected to the user)
        select_incident_value = get_in(
            body,
            "view.state.values.incident_update_select_incident.incident_update_select_incident.initial_option.value",
        )
    if not select_incident_value:
        return False
    try:
        incident_id = int(select_incident_value)
    except ValueError:
        logger.warning(
            f"Select incident ('{select_incident_value}') was not a valid incident ID!"
        )
        return None
    return Incident.objects.get(id=incident_id)

get_user_from_context

get_user_from_context(body: dict[str, Any]) -> User | None

Returns a User or None, from a Slack body.

Source code in src/firefighter/slack/slack_incident_context.py

def get_user_from_context(body: dict[str, Any]) -> User | None:
    """Returns a User or None, from a Slack body."""
    sender_id = get_in(body, "user.id", body.get("user_id"))

    return SlackUser.objects.get_user_by_slack_id(slack_id=sender_id)

slack_templating

date_time cached

date_time(date: datetime | None) -> str

Common format for datetime.

Parameters:

• date (datetime | None) –

  your datetime

Returns:

• str ( str ) –

  datetime in format YYYY-MM-DD HH:MM

Source code in src/firefighter/slack/slack_templating.py

@cache
def date_time(date: datetime | None) -> str:
    """Common format for datetime.

    Args:
        date (datetime | None): your datetime

    Returns:
        str: datetime in format `YYYY-MM-DD HH:MM`
    """
    return localtime(date).strftime("%Y-%m-%d %H:%M")

md_quote_filter

md_quote_filter(val: str | T) -> str | T

Add > on newlines for MD quotes.

Source code in src/firefighter/slack/slack_templating.py

def md_quote_filter(val: str | T) -> str | T:
    """Add > on newlines for MD quotes."""
    if isinstance(val, str):
        return val.replace("\n", "\n> ")
    return val

shorten_long

shorten_long(text: str, width: int, **kwargs: Any) -> str

Shorten text while keeping newlines and most formatting.

Source code in src/firefighter/slack/slack_templating.py

def shorten_long(text: str, width: int, **kwargs: Any) -> str:
    """Shorten text while keeping newlines and most formatting."""
    kwargs.setdefault("placeholder", " [...]")
    kwargs.setdefault("tabsize", 4)
    kwargs.setdefault("max_lines", 1)
    kwargs.setdefault("drop_whitespace", False)
    kwargs.setdefault("replace_whitespace", False)
    kwargs.setdefault("break_long_words", True)
    w = TextWrapper(
        width=width,
        **kwargs,
    )
    return w.fill(text)

user_slack_handle_or_name

user_slack_handle_or_name(user: User | None) -> str

Returns the Slack handle of the user in Slack MD format (<@SLACK_ID>) or the user full name.

Source code in src/firefighter/slack/slack_templating.py

def user_slack_handle_or_name(user: User | None) -> str:
    """Returns the Slack handle of the user in Slack MD format (`<@SLACK_ID>`) or the user full name."""
    if user is None:
        return "∅"

    if hasattr(user, "slack_user") and user.slack_user:
        return f"<@{user.slack_user.slack_id}>"
    return user.full_name

tasks

fetch_conversations_members

fetch_conversations_members_from_slack

fetch_conversations_members_from_slack(
    client: WebClient = DefaultWebClient, queryset: QuerySet[Conversation] | None = None
) -> list[Conversation]

Update the members and metadata of Slack Conversations in DB, from Slack API.

Only fetches conversations that are not IncidentChannels.

Parameters:

• client (WebClient, default: DefaultWebClient ) –

  Slack SDK client. Defaults to DefaultWebClient.

• queryset (Optional[QuerySet[Conversation]], default: None ) –

  Conversation to update. Defaults to None. If None, all applicable

Returns:

• list[Conversation] –

  list[Conversation]: List of conversations that could not be updated.

Raises:

• TypeError –

  If the members list is not a list.

Source code in src/firefighter/slack/tasks/fetch_conversations_members.py

@slack_client
def fetch_conversations_members_from_slack(
    client: WebClient = DefaultWebClient,
    queryset: QuerySet[Conversation] | None = None,
) -> list[Conversation]:
    """Update the members and metadata of Slack Conversations in DB, from Slack API.

    Only fetches conversations that are not IncidentChannels.

    Args:
        client (WebClient, optional): Slack SDK client. Defaults to DefaultWebClient.
        queryset (Optional[QuerySet[Conversation]], optional): Conversation to update. Defaults to None. If None, all applicable

    Returns:
        list[Conversation]: List of conversations that could not be updated.

    Raises:
        TypeError: If the members list is not a list.
    """
    conversations: QuerySet[Conversation] = (
        queryset
        if queryset and queryset.model != Conversation
        else Conversation.objects.not_incident_channel(queryset)
    )

    fails: list[Conversation] = []
    fails = fails + list(conversations.filter(channel_id__isnull=True))

    conversations_members: dict[str, list[str]] = {}
    conversations_info: dict[str, dict[str, Any]] = {}

    for conversation in conversations:
        # Fetch members IDs
        try:
            conversation_members: list[str] = client.conversations_members(
                channel=conversation.channel_id
            ).get("members", [])
        except SlackApiError:
            logger.warning(f"Could not fetch members for {conversation.channel_id}")
            continue
        if not isinstance(conversation_members, list):
            err_msg = f"conversation_members is not a list: {conversation_members}"  # type: ignore[unreachable]
            raise TypeError(err_msg)

        # Fetch conversation info
        conversation_info = client.conversations_info(channel=conversation.channel_id)

        # Save members
        members_slack_ids = conversation_members
        conversations_members[conversation.channel_id] = members_slack_ids

        # Save info (channel_id, channel_name, channel_type, status)
        conversation_data_kwargs_tup = Conversation.objects.parse_slack_response(
            conversation_info
        )
        conversation_data_kwargs = {
            "name": conversation_data_kwargs_tup[1],
            "_type": conversation_data_kwargs_tup[2],
            "_status": conversation_data_kwargs_tup[3],
        }

        conversations_info[conversation.channel_id] = conversation_data_kwargs

    # Get all usergroups members
    all_members_usergroups: set[str] = set(
        functools.reduce(operator.iadd, conversations_members.values(), [])
    )

    all_members_mapping: dict[str, User] = {}

    # Get all users from their Slack IDs
    for member_slack_id in all_members_usergroups:
        user = SlackUser.objects.get_user_by_slack_id(member_slack_id)
        if user is not None:
            all_members_mapping[member_slack_id] = user
            logger.info(user)
            continue

        logger.error(f"Could not retrieve user for Slack ID {member_slack_id}")

    # Usergroups users mapping
    usergroups_members_users: dict[str, list[User]] = {
        k: [all_members_mapping[y] for y in v] for k, v in conversations_members.items()
    }
    logger.debug(usergroups_members_users)

    # Save all usergroups members
    with transaction.atomic():
        for conversation in conversations:
            if not isinstance(conversation.channel_id, str):
                err_msg = f"Conversation {conversation.channel_id} has no channel_id"  # type: ignore[unreachable]
                raise TypeError(err_msg)
            if conversation.channel_id in usergroups_members_users:
                conversation.members.set(
                    usergroups_members_users[conversation.channel_id]
                )
                conversation.__dict__.update(
                    conversations_info[conversation.channel_id]
                )
                conversation.save()
                continue
            fails.append(conversation)
            logger.warning(
                f"Could not save members and info for non existent conversation {conversation.channel_id}"
            )
    return fails

fetch_conversations_members_from_slack_celery

fetch_conversations_members_from_slack_celery(
    *_args: Any, queryset: QuerySet[Conversation] | None = None, **_options: Any
) -> dict[str, list[str]]

Wrapper around the actual task, as Celery doesn't support passing Django models.

Source code in src/firefighter/slack/tasks/fetch_conversations_members.py

@shared_task(
    name="slack.fetch_conversations_members_from_slack",
    retry_kwargs={"max_retries": 2},
    default_retry_delay=90,
)
def fetch_conversations_members_from_slack_celery(
    *_args: Any,
    queryset: QuerySet[Conversation] | None = None,
    **_options: Any,
) -> dict[str, list[str]]:
    """Wrapper around the actual task, as Celery doesn't support passing Django models."""
    failed_conversations = fetch_conversations_members_from_slack(
        queryset=queryset,
    )
    return {"failed_conversations": [x.channel_id for x in failed_conversations]}

reminder_postmortem

publish_postmortem_reminder

publish_postmortem_reminder(incident: Incident, client: WebClient = DefaultWebClient) -> None

XXX Should return a message object, not send it.

Source code in src/firefighter/slack/tasks/reminder_postmortem.py

@slack_client
def publish_postmortem_reminder(
    incident: Incident,
    client: WebClient = DefaultWebClient,
) -> None:
    """XXX Should return a message object, not send it."""
    if not hasattr(incident, "postmortem_for"):
        logger.warning(
            "Trying to send PostMortem reminder for incident #%s with no PostMortem!",
            incident.id,
        )
        return

    if not (hasattr(incident, "conversation") and incident.conversation.channel_id):
        logger.warning(
            "No conversation to post PM reminder for incident {incident.id}."
        )
        return
    update_status_message = SlackMessageIncidentPostMortemReminder(incident)
    incident.conversation.send_message_and_save(
        update_status_message,
        client=client,
        strategy=SlackMessageStrategy.REPLACE,
        strategy_args={
            "replace": (
                SlackMessageIncidentFixedNextActions.id,
                SlackMessageIncidentPostMortemReminder.id,
            )
        },
    )

send_message

send_message

send_message(client: WebClient = DefaultWebClient, *args: Any, **kwargs: Any) -> dict[str, Any] | bytes

Sends a message to a channel. All arguments are passed to the Slack API.

Source code in src/firefighter/slack/tasks/send_message.py

@shared_task(
    name="slack.send_message",
    autoretry_for=(SlackClientError,),
    retry_kwargs={"max_retries": 2},
    default_retry_delay=30,
)
@slack_client
def send_message(  # pylint: disable=keyword-arg-before-vararg
    client: WebClient = DefaultWebClient, *args: Any, **kwargs: Any
) -> dict[str, Any] | bytes:
    """Sends a message to a channel. All arguments are passed to the Slack API."""
    return client.chat_postMessage(*args, **kwargs).data

send_reminders

slack_save_reminder_message

slack_save_reminder_message(message_response_data: SlackResponse, *args: int, **_kwargs: Any) -> bool

Save the firefighter.slack.models.Message from a Slack response. First args is an firefighter.incidents.models.Incident ID.

Parameters:

• message_response_data (dict) –

  SlackResponse data.

• *args (int, default: () ) –

  Expect one value, the firefighter.incidents.models.Incident ID.

• **_kwargs (Any, default: {} ) –

  Ignored.

Source code in src/firefighter/slack/tasks/send_reminders.py

@shared_task(
    name="slack.slack_save_reminder_message",
    retry_kwargs={"max_retries": 5},
    default_retry_delay=30,
)
def slack_save_reminder_message(
    message_response_data: SlackResponse, *args: int, **_kwargs: Any
) -> bool:
    """Save the [firefighter.slack.models.Message][] from a Slack response. First `args` is an [firefighter.incidents.models.Incident][] ID.

    Args:
        message_response_data (dict): SlackResponse data.
        *args: Expect one value, the [firefighter.incidents.models.Incident][] ID.
        **_kwargs: Ignored.
    """
    user = SlackUser.objects.get_user_by_slack_id(
        slack_id=message_response_data["message"]["user"]
    )
    conversation = IncidentChannel.objects.get(
        channel_id=message_response_data["channel"]
    )
    ts = datetime.fromtimestamp(
        float(message_response_data["message"]["ts"]), tz=timezone.utc
    )
    if user is None or not hasattr(user, "slack_user") or user.slack_user is None:
        msg = f"User not found for slack_id {message_response_data['message']['user']}"
        raise ValueError(msg)
    Message(
        conversation=conversation,
        ts=ts,
        type=message_response_data["message"]["type"],
        ff_type=message_response_data["message"]["metadata"]["event_type"],
        user=user.slack_user,
        incident_id=args[0],
    ).save()
    return True

sync_users

sync_users

sync_users(*_args: Any, **_options: Any) -> None

Retrieves users from Slack and updates the database (e.g. name, new profile picture, email, active status...).

Source code in src/firefighter/slack/tasks/sync_users.py

@shared_task(
    name="slack.sync_users",
    retry_kwargs={"max_retries": 2},
    default_retry_delay=90,
)
def sync_users(*_args: Any, **_options: Any) -> None:
    """Retrieves users from Slack and updates the database (e.g. name, new profile picture, email, active status...)."""
    task()

update_usergroups_members

update_usergroups_members_from_slack_celery

update_usergroups_members_from_slack_celery(*args: Any, **options: Any) -> dict[str, list[str | None]]

Wrapper around the actual task, as Celery doesn't support passing Django models.

Source code in src/firefighter/slack/tasks/update_usergroups_members.py

@shared_task(
    name="slack.update_usergroups_members_from_slack",
    retry_kwargs={"max_retries": 2},
    default_retry_delay=90,
)
def update_usergroups_members_from_slack_celery(
    *args: Any,
    **options: Any,
) -> dict[str, list[str | None]]:
    """Wrapper around the actual task, as Celery doesn't support passing Django models."""
    fails = update_usergroups_members_from_slack(*args, **options)
    return {"failed_groups": [x.usergroup_id for x in fails]}

update_users

update_users_from_slack

update_users_from_slack(*_args: Any, **_options: Any) -> None

Retrieves users from Slack and updates the database (e.g. name, new profile picture, email, active status...).

Source code in src/firefighter/slack/tasks/update_users.py

@shared_task(
    name="slack.update_users_from_slack",
    retry_kwargs={"max_retries": 2},
    default_retry_delay=90,
)
def update_users_from_slack(*_args: Any, **_options: Any) -> None:
    """Retrieves users from Slack and updates the database (e.g. name, new profile picture, email, active status...)."""
    queryset = SlackUser.objects.all().order_by("user__updated_at")[:100]
    for slack_user in queryset:
        slack_user.update_user_info()

urls

utils

channel_name_from_incident

channel_name_from_incident(incident: Incident) -> str

Lowercase, truncated at 80 chars, this is obviously the channel #name.

Source code in src/firefighter/slack/utils.py

def channel_name_from_incident(incident: Incident) -> str:
    """Lowercase, truncated at 80 chars, this is obviously the channel #name."""
    if (
        not hasattr(incident, "created_at")
        or not hasattr(incident, "id")
        or not incident.created_at
        or not incident.id
    ):
        raise RuntimeError(
            "Incident must be saved before slack_channel_name can be computed"
        )
    date_formatted = localtime(incident.created_at).strftime("%Y%m%d")
    if incident.environment is not None and incident.environment.value != "PRD":
        topic = f"{date_formatted}-{str(incident.id)[:8]}-{incident.environment.value}-{incident.component.name}"
    else:
        topic = f"{date_formatted}-{str(incident.id)[:8]}-{incident.component.name}"

    # Strip non-alphanumeric characters, cut at 80 chars
    # XXX django.utils.text.slugify should be used instead
    topic = topic.replace(" ", "-")
    topic = NON_ALPHANUMERIC_CHARACTERS.sub("-", topic)
    return topic.lower()[:80]

get_slack_user_id_from_body

get_slack_user_id_from_body(body: dict[str, Any]) -> str | None

Get the slack user id from the body of a Slack request, in user_id or user.id.

Source code in src/firefighter/slack/utils.py

def get_slack_user_id_from_body(body: dict[str, Any]) -> str | None:
    """Get the slack user id from the body of a Slack request, in `user_id` or `user.id`."""
    return body.get("user_id", body.get("user", {}).get("id"))

respond

respond(
    body: dict[str, Any], text: str = "", blocks: str | Sequence[dict[str, Any] | Block] | None = None, client: WebClient = DefaultWebClient
) -> None

Respond to the user, depending on where the message was coming from.

Source code in src/firefighter/slack/utils.py

@slack_client
def respond(
    body: dict[str, Any],
    text: str = "",
    blocks: str | Sequence[dict[str, Any] | Block] | None = None,
    client: WebClient = DefaultWebClient,
) -> None:
    """Respond to the user, depending on where the message was coming from."""
    user_id: str | None = body.get("user_id", body.get("user", {}).get("id"))
    channel_id: str | None = body.get("channel_id", body.get("channel", {}).get("id"))

    if not user_id and not channel_id:
        raise ValueError(
            "Cannot find user_id (or user.id) or channel_id (or channel.id) in the body. At least one is required."
        )

    # From Direct Message => Always respond on the conv between the user and the bot.
    if (body.get("channel_name") == "directmessage" or not channel_id) and user_id:
        # We should always be able to respond in this conversation...
        client.chat_postMessage(channel=user_id, text=text, blocks=blocks)
        return
    if not user_id:
        raise ValueError("Cannot find user_id (or user.id) in the body.")

    # From channel => post as ephemeral in the channel
    try:
        _send_ephemeral(text, blocks, client, user_id, channel_id)

    except (SlackApiError, ValueError):
        logger.warning(
            "Failed to send ephemeral chat message to user! Body: %s",
            body,
            exc_info=True,
        )
        # Fallback to DM
        if body.get("type") != "view_submission":
            client.chat_postMessage(
                channel=user_id,
                text=":warning: The bot could not respond in the channel you invoked it. Please add it to this channel or conversation if you want to interact with the bot there. If you believe this is a bug, please tell @pulse.",
            )
        client.chat_postMessage(channel=user_id, text=text, blocks=blocks)

views

events

handle_message_events_ignore

handle_message_events_ignore() -> None

Ignore other message events. Must be the last event handler for message.

Source code in src/firefighter/slack/views/events/__init__.py

@app.event("message")
def handle_message_events_ignore() -> None:
    """Ignore other message events.
    Must be the last event handler for message.
    """
    return

actions_and_shortcuts

open_link

open_link(ack: Ack) -> None

Does nothing. ack() is mandatory, even on buttons that open a URL.

Source code in src/firefighter/slack/views/events/actions_and_shortcuts.py

@app.action("open_link")
def open_link(ack: Ack) -> None:
    """Does nothing. ack() is mandatory, even on buttons that open a URL."""
    ack()

update_update_modal

update_update_modal(ack: Ack, body: dict[str, Any]) -> None

Reacts to the selection of an incident, in the select modal.

Source code in src/firefighter/slack/views/events/actions_and_shortcuts.py

@app.action("incident_update_select_incident")
def update_update_modal(ack: Ack, body: dict[str, Any]) -> None:
    """Reacts to the selection of an incident, in the select modal."""
    ack()
    logger.info(body)

    callback_id = get_in(body, "view.callback_id")
    if callback_id in selectable:
        view = selectable[callback_id].build_modal_with_context(
            body, callback_id=callback_id
        )
        update_modal(body=body, view=view)

channel_archive

channel_id_changed

channel_rename

channel_shared

channel_unarchive

channel_unshared

commands

SLACK_BUILTIN_COMMANDS module-attribute

SLACK_BUILTIN_COMMANDS = (
    "/archive",
    "/call",
    "/collapse",
    "/dm",
    "/expand",
    "/feed",
    "/invite",
    "/leave",
    "/msg",
    "/remind",
    "/remove",
    "/rename",
    "/search",
    "/shrug",
    "/status",
    "/topic",
)

List of all Slack built-in commands (https://slack.com/help/articles/201259356 checked August 2022)

register_commands

register_commands() -> None

Register the command with its aliases. Commands are checked: - Fix commands that does not start with a slash, contain spaces, uppercase characters or are longer than 32 characters. - Ignore commands which names are built-in Slack commands.

⚠️ Don't forget to add the command and its aliases on your Slack App settings.

Source code in src/firefighter/slack/views/events/commands.py

def register_commands() -> None:
    """Register the command with its aliases.
    Commands are checked:
    - Fix commands that does not start with a slash, contain spaces, uppercase characters or are longer than 32 characters.
    - Ignore commands which names are built-in Slack commands.

    ⚠️ Don't forget to add the command and its aliases on your Slack App settings.
    """
    command_main: str = settings.SLACK_INCIDENT_COMMAND
    command_aliases: list[str] = settings.SLACK_INCIDENT_COMMAND_ALIASES

    commands = [command_main, *command_aliases]
    logger.debug(f"Registered commands: {commands}")
    for command in commands:
        if not command.startswith("/"):
            command = f"/{command}"  # noqa: PLW2901
            logger.warning(
                f"Command '{command}' does not start with a slash. We added one but please fix your configuration."
            )
        if " " in command:
            command = command.replace(" ", "")  # noqa: PLW2901
            logger.warning(
                f"Command '{command}' contained spaces. We removed them but please fix your configuration."
            )
        if len(command) > 32:
            command = command[:32]  # noqa: PLW2901
            logger.warning(
                f"Command '{command}' was longer than 32 characters. We truncated it but please fix your configuration."
            )
        if any(char.isupper() for char in command):
            command = command.lower()  # noqa: PLW2901
            logger.warning(
                f"Command '{command}' contained uppercase characters. We lower-cased them but please fix your configuration."
            )
        if command in SLACK_BUILTIN_COMMANDS:
            logger.warning(
                f"Command '{command}' is a built-in command, skipping registration. This command will not work."
            )
            continue
        app.command(command)(manage_incident)

home

member_joined_channel

member_joined_channel

member_joined_channel(event: dict[str, Any]) -> None

When a user joins a channel, we add it to the list of conversations.

API Reference: https://api.slack.com/events/member_joined_channel

Source code in src/firefighter/slack/views/events/member_joined_channel.py

@app.event("member_joined_channel")
def member_joined_channel(event: dict[str, Any]) -> None:
    """When a user joins a channel, we add it to the list of conversations.

    API Reference: https://api.slack.com/events/member_joined_channel
    """
    logger.debug(event)
    channel_id = get_in(event, "channel")
    channel_type = get_in(event, "channel_type")
    user_id = get_in(event, "user")
    inviter = get_in(event, "inviter")
    if not channel_id:
        logger.warning(f"Invalid event! {event}")
        return

    conversation = Conversation.objects.get_or_none(channel_id=channel_id)
    if not conversation:
        logger.warning(f"Conversation {channel_id} does not exist!")
        return

    if channel_type == "C" and conversation.type != ConversationType.PUBLIC_CHANNEL:
        conversation.type = ConversationType.PUBLIC_CHANNEL
        conversation.save()
    elif channel_type == "G" and conversation.type != ConversationType.PRIVATE_CHANNEL:
        conversation.type = ConversationType.PRIVATE_CHANNEL
        conversation.save()
    if user_id:
        user = SlackUser.objects.get_user_by_slack_id(slack_id=user_id)
        if user is None:
            logger.warning(f"User {user_id} does not exist!")
            return
        conversation.members.add(user)

        # Check if Conversation is also IncidentChannel
        try:
            incident_channel: IncidentChannel = conversation.incidentchannel
        except IncidentChannel.DoesNotExist:
            return

        if inviter and incident_channel:
            inviter_user = SlackUser.objects.get_user_by_slack_id(slack_id=inviter)
            if inviter_user is None:
                logger.warning(f"Inviter {inviter} does not exist!")
                return

            logger.debug(f"User {user.id} joined {conversation} by {inviter_user.id}")

        else:
            inviter_user = None

        incident_channel.members.add(user)

member_left_channel

member_left_channel

member_left_channel(event: dict[str, Any]) -> None

When a user leaves a channel, we remove it from the list of conversations.

API Reference: https://api.slack.com/events/member_left_channel

Source code in src/firefighter/slack/views/events/member_left_channel.py

@app.event("member_left_channel")
def member_left_channel(event: dict[str, Any]) -> None:
    """When a user leaves a channel, we remove it from the list of conversations.

    API Reference: https://api.slack.com/events/member_left_channel
    """
    channel_id = get_in(event, "channel")
    channel_type = get_in(event, "channel_type")
    user_id = get_in(event, "user")
    if not channel_id:
        logger.warning(f"Invalid event! {event}")
        return

    conversation = Conversation.objects.get_or_none(channel_id=channel_id)
    if not conversation:
        logger.warning(f"Conversation {channel_id} does not exist!")
        return

    if channel_type == "C" and conversation.type != ConversationType.PUBLIC_CHANNEL:
        conversation.type = ConversationType.PUBLIC_CHANNEL
        conversation.save()
    elif channel_type == "G" and conversation.type != ConversationType.PRIVATE_CHANNEL:
        conversation.type = ConversationType.PRIVATE_CHANNEL
        conversation.save()
    if user_id:
        user = SlackUser.objects.get_user_by_slack_id(slack_id=user_id)
        if user is None:
            logger.warning(f"User {user_id} does not exist!")
            return
        conversation.members.remove(user)

        # Check if Conversation is also IncidentChannel
        try:
            incident_channel: IncidentChannel = conversation.incidentchannel
        except IncidentChannel.DoesNotExist:
            return
        if incident_channel:
            logger.info(f"User {user} left {conversation} ")

            incident = incident_channel.incident
            IncidentMembership.objects.filter(incident=incident, user=user).delete()

message

handle_message_events_ignore

handle_message_events_ignore() -> None

Ignore other message events. Must be the last event handler for message.

Source code in src/firefighter/slack/views/events/message.py

@app.event({"type": "message", "subtype": "channel_topic"})
def handle_message_events_ignore() -> None:
    """Ignore other message events.
    Must be the last event handler for message.
    """
    return

message_deleted

reaction_added

reaction_added_ignore

reaction_added_ignore() -> None

Ignore other reaction_added events. Must be the last event handler for reaction_added.

Source code in src/firefighter/slack/views/events/reaction_added.py

@app.event("reaction_added")
def reaction_added_ignore() -> None:
    """Ignore other reaction_added events.
    Must be the last event handler for reaction_added.
    """
    return

modals

base_modal

base

MessageForm

MessageForm()

Bases: SlackModal, Generic[T]

Form wrapper to use a Django form in a Slack message.

Provided a Django form in form_class, it will handle: - generation of Slack Blocks - validation and submission of the form

Source code in src/firefighter/slack/views/modals/base_modal/base.py

def __init__(self) -> None:
    if self.handle_modal_fn is not None:
        self.handler_fn_args = inspect.getfullargspec(self.handle_modal_fn).args
        if len(self.handler_fn_args) > 0 and self.handler_fn_args[0] in {
            "self",
            "cls",
        }:
            self.handler_fn_args.pop(0)

    self.builder_fn_args = inspect.getfullargspec(self.build_modal_fn).args
    if len(self.builder_fn_args) > 0 and self.builder_fn_args[0] in {"self", "cls"}:
        self.builder_fn_args.pop(0)

    self._register_actions_shortcuts()

callback_id instance-attribute

callback_id: str | re.Pattern[str]

Callback ID for the Slack View

open_shortcut class-attribute instance-attribute

open_shortcut: str | None = None

Slack shortcut to open the modal.

ModalForm

ModalForm()

Bases: SlackModal, Generic[T]

Specific SlackModal to handle a Django form.

Provided a Django form in form_class, it will handle: - generation of Slack Blocks - validation and submission of the form

Source code in src/firefighter/slack/views/modals/base_modal/base.py

def __init__(self) -> None:
    if self.handle_modal_fn is not None:
        self.handler_fn_args = inspect.getfullargspec(self.handle_modal_fn).args
        if len(self.handler_fn_args) > 0 and self.handler_fn_args[0] in {
            "self",
            "cls",
        }:
            self.handler_fn_args.pop(0)

    self.builder_fn_args = inspect.getfullargspec(self.build_modal_fn).args
    if len(self.builder_fn_args) > 0 and self.builder_fn_args[0] in {"self", "cls"}:
        self.builder_fn_args.pop(0)

    self._register_actions_shortcuts()

callback_id instance-attribute

callback_id: str | re.Pattern[str]

Callback ID for the Slack View

open_shortcut class-attribute instance-attribute

open_shortcut: str | None = None

Slack shortcut to open the modal.

SlackModal

SlackModal()

Two main responsibilities: - Register the modal on Slack to open it with shortcuts or actions - Provide useful context for build_modal_fn and handle_modal_fn, such as body, incident, user, etc.

Source code in src/firefighter/slack/views/modals/base_modal/base.py

def __init__(self) -> None:
    if self.handle_modal_fn is not None:
        self.handler_fn_args = inspect.getfullargspec(self.handle_modal_fn).args
        if len(self.handler_fn_args) > 0 and self.handler_fn_args[0] in {
            "self",
            "cls",
        }:
            self.handler_fn_args.pop(0)

    self.builder_fn_args = inspect.getfullargspec(self.build_modal_fn).args
    if len(self.builder_fn_args) > 0 and self.builder_fn_args[0] in {"self", "cls"}:
        self.builder_fn_args.pop(0)

    self._register_actions_shortcuts()

callback_id instance-attribute

callback_id: str | re.Pattern[str]

Callback ID for the Slack View

open_shortcut class-attribute instance-attribute

open_shortcut: str | None = None

Slack shortcut to open the modal.

form_utils

SafeOption

SafeOption(
    *,
    value: str,
    label: str | None = None,
    text: str | dict[str, Any] | TextObject | None = None,
    description: str | dict[str, Any] | TextObject | None = None,
    url: str | None = None,
    **others: dict[str, Any]
)

Bases: Option

Make sure we are creating valid Option, warn otherwise.

Source code in src/firefighter/slack/views/modals/base_modal/form_utils.py

def __init__(
    self,
    *,
    value: str,
    label: str | None = None,
    text: str | dict[str, Any] | TextObject | None = None,  # Block Kit
    description: str | dict[str, Any] | TextObject | None = None,
    url: str | None = None,
    **others: dict[str, Any],
) -> None:
    if len(value) > 75:
        logger.warning("Option value is too long: %s", value)
        value = value[:75]
    if label and len(label) > 75:
        logger.warning("Option label is too long: %s", label)
        label = label[:74] + "…"
    super().__init__(
        value=value,
        label=label,
        text=text,
        description=description,
        url=url,
        **others,
    )

SlackForm

SlackForm(form: type[T], slack_fields: SlackFormAttributesDict | None = None)

Bases: Generic[T]

Source code in src/firefighter/slack/views/modals/base_modal/form_utils.py

def __init__(
    self,
    form: type[T],
    slack_fields: SlackFormAttributesDict | None = None,
):
    self.form_class = form

    if hasattr(form, "slack_fields") and form.slack_fields:  # type: ignore
        self.slack_fields = form.slack_fields  # type: ignore
    else:
        self.slack_fields = slack_fields or {}

slack_blocks

slack_blocks(block_wrapper: Literal['section_accessory', 'input', 'action'] = 'input') -> list[Block]

Return the list of blocks from a SlackForm.

Parameters:

• block_wrapper (str, default: 'input' ) –

  Type of blocks destination. Defaults to "input".

Raises:

• ValueError –

  Raised if one of the initial value does not match the field type.

Returns:

• list[Block] –

  list[Block]: The List of Slack Blocks.

Source code in src/firefighter/slack/views/modals/base_modal/form_utils.py

def slack_blocks(
    self,
    block_wrapper: Literal["section_accessory", "input", "action"] = "input",
) -> list[Block]:
    """Return the list of blocks from a SlackForm.

    Args:
        block_wrapper (str, optional): Type of blocks destination. Defaults to "input".

    Raises:
        ValueError: Raised if one of the initial value does not match the field type.

    Returns:
        list[Block]: The List of Slack Blocks.
    """
    blocks = []
    for field_name, f in self.form.fields.items():
        # Get the default data, from the form data, the field.initial or the form.initial
        f.initial = self.get_field_initial(field_name, f)

        slack_input_kwargs: dict[str, Any] = {}
        slack_block_kwargs: dict[str, Any] = {}

        # Set Field common args
        slack_block_kwargs["label"] = (f.label or field_name.title())[:2000]
        slack_block_kwargs["hint"] = f.help_text[:2000]
        slack_block_kwargs["optional"] = not f.required

        # Set custom Slack SDK fields
        post_block, pre_block = self._parse_field_slack_args(
            field_name, f, slack_input_kwargs, slack_block_kwargs
        )

        slack_input_element = self._get_input_element(
            f, field_name, slack_input_kwargs
        )
        if slack_input_element is None:
            continue
        blocks += self._wrap_field_in_block(
            block_wrapper,
            field_name,
            f,
            slack_block_kwargs,
            slack_input_element,
            post_block,
            pre_block,
        )
    return blocks

SlackFormJSONEncoder

Bases: JSONEncoder

JSON encoder that can handle UUIDs and Django models. Used to serialize the form data to JSON for Slack modal private_metadata.

slack_view_submission_to_dict

slack_view_submission_to_dict(body: dict[str, Any | str]) -> dict[str, str | Any]

Returns a dict of the form data from a Slack view submission.

Source code in src/firefighter/slack/views/modals/base_modal/form_utils.py

def slack_view_submission_to_dict(
    body: dict[str, Any | str],
) -> dict[str, str | Any]:
    """Returns a dict of the form data from a Slack view submission."""
    if body.get("view"):
        path = ["view", "state", "values"]
    elif body.get("type") == "block_actions":
        path = ["state", "values"]
    else:
        logger.warning("Unknown Slack view submission format: %s", body)
        path = ["view", "state", "values"]

    values: dict[str, dict[str, Any]] = get_in(body, path)
    data: dict[str, str | Any] = {}
    if not isinstance(values, dict):
        raise TypeError("Expected a values dict in the body")

    # We expect only one action per input
    # The action_id must be block_id or block_id___{whatever}
    # We support block_id and block_id___{whatever} because SLack won't update the value in the user's form if the action_id is the same
    # Hence, we need to add a unique identifier to the action_id to force Slack to update the value, replacing the input by another one
    for block in values.values():
        if len(block) == 0:
            continue
        action_id: str = next(iter(block.keys()))
        action_id_stripped = action_id.split("___", 1)[0].strip()
        input_field = block.get(action_id)
        if not isinstance(input_field, dict):
            raise TypeError("Expected input_field to be a dict")

        if input_field.get("type") == "plain_text_input":
            data[action_id_stripped] = input_field.get("value")
        elif input_field.get("type") == "datetimepicker":
            data[action_id_stripped] = (
                datetime.fromtimestamp(
                    cast("float", input_field.get("selected_date_time")),
                    tz=TZ,
                )
                if input_field.get("selected_date_time")
                else None
            )
        elif input_field.get("type") == "static_select":
            data[action_id_stripped] = get_in(input_field, ["selected_option", "value"])
        elif input_field.get("type") == "users_select":
            user_id = get_in(
                input_field,
                [
                    "selected_user",
                ],
            )
            user_obj = (
                SlackUser.objects.get_user_by_slack_id(slack_id=user_id)
                if user_id
                else None
            )
            data[action_id_stripped] = user_obj.id if user_obj else None
    return data

mixins

modal_utils

close

CloseModal

CloseModal()

Bases: IncidentSelectableModalMixin, ModalForm[CloseIncidentFormSlack]

Source code in src/firefighter/slack/views/modals/base_modal/base.py

def __init__(self) -> None:
    if self.handle_modal_fn is not None:
        self.handler_fn_args = inspect.getfullargspec(self.handle_modal_fn).args
        if len(self.handler_fn_args) > 0 and self.handler_fn_args[0] in {
            "self",
            "cls",
        }:
            self.handler_fn_args.pop(0)

    self.builder_fn_args = inspect.getfullargspec(self.build_modal_fn).args
    if len(self.builder_fn_args) > 0 and self.builder_fn_args[0] in {"self", "cls"}:
        self.builder_fn_args.pop(0)

    self._register_actions_shortcuts()

handle_modal_fn

handle_modal_fn(ack: Ack, body: dict[str, Any], incident: Incident, user: User) -> None | bool

Handle response from /incident close modal.

Source code in src/firefighter/slack/views/modals/close.py

def handle_modal_fn(  # type: ignore[override]
    self, ack: Ack, body: dict[str, Any], incident: Incident, user: User
) -> None | bool:
    """Handle response from /incident close modal."""
    slack_form = self.handle_form_errors(
        ack, body, forms_kwargs={"initial": self._get_initial_form_values(incident)}
    )
    if slack_form is None:
        return None
    form = slack_form.form
    # If fields haven't changed, don't include them in the update.
    update_kwargs = {}
    for changed_key in form.changed_data:
        if changed_key == "component":
            update_kwargs["component_id"] = form.cleaned_data[changed_key].id
        if changed_key in {"description", "title", "message"}:
            update_kwargs[changed_key] = form.cleaned_data[changed_key]
    # Check can close
    can_close, reasons = incident.can_be_closed
    if not can_close:
        logger.warning(
            f"Tried to close an incident that can't be closed yet! Aborting. Incident #{incident.id}. Reasons: {reasons}"
        )
        respond(
            body=body,
            text=f"It looks like this incident #{incident.id} could not be closed.\nReasons: {reasons}.\nPlease tell @pulse (#tech-pe-pulse) if you think this is an error.",
        )
        return False
    self._trigger_incident_workflow(incident, user, update_kwargs)
    return None

downgrade_workflow

key_event_message

KeyEvents

KeyEvents()

Bases: MessageForm[IncidentUpdateKeyEventsForm]

Source code in src/firefighter/slack/views/modals/base_modal/base.py

def __init__(self) -> None:
    if self.handle_modal_fn is not None:
        self.handler_fn_args = inspect.getfullargspec(self.handle_modal_fn).args
        if len(self.handler_fn_args) > 0 and self.handler_fn_args[0] in {
            "self",
            "cls",
        }:
            self.handler_fn_args.pop(0)

    self.builder_fn_args = inspect.getfullargspec(self.build_modal_fn).args
    if len(self.builder_fn_args) > 0 and self.builder_fn_args[0] in {"self", "cls"}:
        self.builder_fn_args.pop(0)

    self._register_actions_shortcuts()

open_shortcut class-attribute instance-attribute

open_shortcut: str | None = None

Slack shortcut to open the modal.

handle_modal_fn

handle_modal_fn(ack: Ack, body: dict[str, Any], user: User, incident: Incident) -> None

Handle the time and date inputs for the key events.

Source code in src/firefighter/slack/views/modals/key_event_message.py

def handle_modal_fn(  # type: ignore[override]
    self, ack: Ack, body: dict[str, Any], user: User, incident: Incident
) -> None:
    """Handle the time and date inputs for the key events."""
    logger.debug(body)

    slack_form: SlackForm[
        IncidentUpdateKeyEventsForm
    ] | None = self.handle_form_errors(
        ack,
        body,
        forms_kwargs={
            "incident": incident,
            "user": user,
        },
    )
    form = slack_form.form if slack_form else None

    if form is None:
        logger.warning("Form is None, skipping save")
        return
    if len(form.errors) > 0:
        self.update_with_form()
        return
    self.form = form
    self.form.save()
    incident.compute_metrics()

    self.update_with_form()

open

OpenModal

OpenModal()

Bases: SlackModal

Source code in src/firefighter/slack/views/modals/base_modal/base.py

def __init__(self) -> None:
    if self.handle_modal_fn is not None:
        self.handler_fn_args = inspect.getfullargspec(self.handle_modal_fn).args
        if len(self.handler_fn_args) > 0 and self.handler_fn_args[0] in {
            "self",
            "cls",
        }:
            self.handler_fn_args.pop(0)

    self.builder_fn_args = inspect.getfullargspec(self.build_modal_fn).args
    if len(self.builder_fn_args) > 0 and self.builder_fn_args[0] in {"self", "cls"}:
        self.builder_fn_args.pop(0)

    self._register_actions_shortcuts()

get_details_modal_form_class staticmethod

get_details_modal_form_class(open_incident_context: OpeningData, incident_type_value: str | None) -> type[SetIncidentDetails[Any]] | None

Get the details modal form class based on the incident type.

Returns None if no incident type is selected.

Source code in src/firefighter/slack/views/modals/open.py

@staticmethod
def get_details_modal_form_class(
    open_incident_context: OpeningData,
    incident_type_value: str | None,
) -> type[SetIncidentDetails[Any]] | None:
    """Get the details modal form class based on the incident type.

    Returns None if no incident type is selected.
    """
    response_type = open_incident_context.get("response_type")
    if response_type is None:
        return None
    incident_types = INCIDENT_TYPES.get(response_type)
    if incident_types and len(incident_types) == 1:
        return incident_types[next(iter(incident_types.keys()))].get("slack_form")
    if incident_types and incident_type_value is not None:
        return incident_types[incident_type_value].get("slack_form")
    logger.debug(
        f"No incident type found for {open_incident_context}. No fallback."
    )
    return None

handle_modal_fn

handle_modal_fn(ack: Ack, body: dict[str, Any], user: User)

Handle response from /incident open modal.

Source code in src/firefighter/slack/views/modals/open.py

def handle_modal_fn(  # type: ignore
    self,
    ack: Ack,
    body: dict[str, Any],
    user: User,
):
    """Handle response from /incident open modal."""
    data: OpeningData = json.loads(body["view"]["private_metadata"])

    details_form_data_raw = data.get("details_form_data", {})

    incident_type_value: str | None = data.get("incident_type", None)

    details_form_modal_class = self.get_details_modal_form_class(
        data, incident_type_value
    )
    if details_form_modal_class:
        details_form_class: type[
            CreateIncidentFormBase
        ] = details_form_modal_class.form_class
        if details_form_class:
            details_form: CreateIncidentFormBase = details_form_class(
                details_form_data_raw
            )
            details_form.is_valid()
            ack()
            try:
                if hasattr(details_form, "trigger_incident_workflow") and callable(
                    details_form.trigger_incident_workflow
                ):
                    details_form.trigger_incident_workflow(
                        creator=user,
                        impacts_data=data.get("impact_form_data") or {},
                    )
            except:  # noqa: E722
                logger.exception("Error triggering incident workflow")

opening

check_current_incidents

details

critical

select_impact

SelectImpactModal

SelectImpactModal()

Bases: IncidentSelectableModalMixin, ModalForm[SelectImpactFormSlack]

TODO: The detailed impacts selected should be saved on the incident.

Source code in src/firefighter/slack/views/modals/base_modal/base.py

def __init__(self) -> None:
    if self.handle_modal_fn is not None:
        self.handler_fn_args = inspect.getfullargspec(self.handle_modal_fn).args
        if len(self.handler_fn_args) > 0 and self.handler_fn_args[0] in {
            "self",
            "cls",
        }:
            self.handler_fn_args.pop(0)

    self.builder_fn_args = inspect.getfullargspec(self.build_modal_fn).args
    if len(self.builder_fn_args) > 0 and self.builder_fn_args[0] in {"self", "cls"}:
        self.builder_fn_args.pop(0)

    self._register_actions_shortcuts()

set_details

SetIncidentDetails

SetIncidentDetails()

Bases: ModalForm[T], Generic[T]

Source code in src/firefighter/slack/views/modals/opening/set_details.py

def __init__(self) -> None:
    if hasattr(self, "id"):
        self.push_action = f"push_{self.id}"
    super().__init__()

open_shortcut class-attribute instance-attribute

open_shortcut: str | None = None

Slack shortcut to open the modal.

handle_modal_fn

handle_modal_fn(ack: Ack, body: dict[str, Any], user: User | None = None)

Handle response from /incident open modal.

Source code in src/firefighter/slack/views/modals/opening/set_details.py

def handle_modal_fn(  # type: ignore
    self,
    ack: Ack,
    body: dict[str, Any],
    user: User | None = None,
):
    """Handle response from /incident open modal."""
    private_metadata: dict[str, Any] = json.loads(
        body.get("view", {}).get("private_metadata", {})
    )
    priority = private_metadata.get("details_form_data", {}).get("priority", None)
    if priority is not None:
        priority = Priority.objects.get(pk=priority)

    slack_form = self.get_form_class()(
        data={**slack_view_submission_to_dict(body), "priority": priority}
    )
    form: T = slack_form.form
    if form.is_valid():
        ack()
        skip_form = False
    else:
        ack(
            response_action="errors",
            errors={
                k: self._concat_validation_errors_msg(v)
                for k, v in form.errors.as_data().items()
            },
        )
        skip_form = True

    if skip_form:
        return
    # ruff: noqa: PLC0415
    from firefighter.slack.views.modals.open import modal_open

    if "priority" in private_metadata and isinstance(
        private_metadata["priority"], str
    ):
        private_metadata["priority"] = Priority.objects.get(
            pk=private_metadata["priority"]
        )
    data = OpeningData(
        details_form_data=cast(dict[str, Any], form.data),
        impact_form_data=private_metadata.get("impact_form_data"),
        incident_type=private_metadata.get("incident_type"),
        response_type=private_metadata.get("response_type"),
        priority=private_metadata.get("priority"),
    )
    view = modal_open.build_modal_fn(open_incident_context=data, user=user)
    view.private_metadata = json.dumps(data, cls=SlackFormJSONEncoder)

    update_modal(
        view=view,
        trigger_id=get_in(body, ["trigger_id"]),
        view_id=get_in(body, ["view", "root_view_id"]),
    )

types

postmortem

select

send_sos

status

trigger_oncall

OnCallModal

OnCallModal()

Bases: IncidentSelectableModalMixin, SlackModal

Source code in src/firefighter/slack/views/modals/base_modal/base.py

def __init__(self) -> None:
    if self.handle_modal_fn is not None:
        self.handler_fn_args = inspect.getfullargspec(self.handle_modal_fn).args
        if len(self.handler_fn_args) > 0 and self.handler_fn_args[0] in {
            "self",
            "cls",
        }:
            self.handler_fn_args.pop(0)

    self.builder_fn_args = inspect.getfullargspec(self.build_modal_fn).args
    if len(self.builder_fn_args) > 0 and self.builder_fn_args[0] in {"self", "cls"}:
        self.builder_fn_args.pop(0)

    self._register_actions_shortcuts()

open_shortcut class-attribute instance-attribute

open_shortcut: str | None = None

Slack shortcut to open the modal.

build_modal_fn

build_modal_fn(incident: Incident, **kwargs: Any) -> View

XXX Should get an incident ID instead of an incident.

Source code in src/firefighter/slack/views/modals/trigger_oncall.py

def build_modal_fn(self, incident: Incident, **kwargs: Any) -> View:
    """XXX Should get an incident ID instead of an incident."""
    previous_pd_incidents = PagerDutyIncident.objects.filter(
        incident_id=incident.id
    )

    pd_services = PagerDutyService.objects.exclude(
        pagerdutyincident__in=previous_pd_incidents,
    ).filter(ignore=False)
    pd_options = [Option(value=str(s.id), label=s.summary) for s in pd_services]

    blocks: list[Block] = [
        SectionBlock(
            text=f"You are about to trigger on-call for incident #{incident.id}. Please select the on-call line that you want to trigger."
        ),
        DividerBlock(),
        InputBlock(
            block_id="oncall_service",
            label="Select on-call line",
            element=RadioButtonsElement(
                action_id="select_oncall_service", options=pd_options
            ),
        )
        if len(pd_options) > 0
        else SectionBlock(
            text=":warning: No PagerDuty services in the database! :warning:\nAdministrator action is needed."
        ),
    ]

    if len(previous_pd_incidents) > 0:
        already_existing_text = [
            f"- <{s.service.web_url}|{s.service.summary}>: <{s.web_url}|_{s.summary}_>\n"
            for s in previous_pd_incidents
        ]
        blocks.extend(
            (
                DividerBlock(),
                SectionBlock(
                    text=f":warning: There are already some on-call lines that have been triggered and can not be triggered again:\n {''.join(already_existing_text)}"
                ),
            )
        )

    return View(
        type="modal",
        title="Trigger on-call"[:24],
        submit="Trigger on-call"[:24] if len(pd_options) > 0 else None,
        callback_id=self.callback_id,
        private_metadata=str(incident.id),
        blocks=blocks,
    )

update

update_roles

update_status

views

SlackEventsHandler

Bases: View

Handle all Slack events.