Skip to content

RAID Module¤

The RAID module provides enhanced Jira integration for incident management and external ticket creation.

Overview¤

RAID (Request And Issue Database) extends FireFighter's capabilities with:

  • Automatic Jira ticket creation for incidents
  • External API for ticket creation (Landbot integration)
  • Smart user resolution based on email domains
  • Priority mapping from P1-P5 to Jira priorities
  • Attachment handling for external submissions
  • Slack notifications for ticket updates

Priority Mapping¤

The RAID module supports full P1-P5 priority mapping:

Priority Level Jira Priority Use Case
P1 Critical 1 System outages, complete service failures
P2 High 2 Major functionality impaired, significant impact
P3 Medium 3 Minor functionality affected, moderate impact
P4 Low 4 Small issues, minimal impact
P5 Lowest 5 Cosmetic issues, enhancement requests

User Resolution¤

The module handles user resolution for external ticket creation:

  1. Internal users: Direct mapping from email to existing users
  2. External users: Email domain-based routing to default Jira users
  3. Slack fallback: Attempts to find users via Slack integration
  4. Default assignment: Falls back to configured default Jira user

Authentication¤

The external ticket creation endpoint (POST /api/v2/firefighter/raid/jira_bot) requires Bearer token authentication.

Each caller must use a dedicated APIToken, generated from the Django back-office at /admin/api/apitokenproxy/. Requests must include the header:

Authorization: Bearer <token>

Common pitfall: HTML formatting in copy-pasted tokens¤

The Django admin renders the token with text styling inherited from rest_framework.authtoken. A regular copy-paste from the admin into a tool that accepts rich-text input (Landbot Webhook header values, some no-code platforms, Notion, etc.) silently carries the styling as inline HTML. The receiving tool displays the value as plain text, but the actual outgoing header looks like:

Authorization: <h2>Bearer <token></h2>

The endpoint then returns 401 not_authenticated because DRF cannot parse the keyword.

Workaround: when copying a token from the admin, paste with Cmd+Shift+V (paste without formatting) or type the token by hand.

Debug tip: if a third-party integration fails with not_authenticated while curl with the same token succeeds, route the integration's webhook to webhook.site to inspect the actual Authorization header it emits.

API Reference¤

raid ¤

RAID is ManoMano's internal issue dispatcher.

It works with Slack and Jira to create, update and close issues.

It is currently not configurable.

Modules:

client ¤

Classes:

RaidJiraClient ¤

RaidJiraClient()

Bases: JiraClient

Methods:

Source code in src/firefighter/jira_app/client.py
def __init__(self) -> None:
    self.url = RAID_JIRA_API_URL

add_attachments_to_issue staticmethod ¤

add_attachments_to_issue(issue_id: str | int, urls: list[str]) -> None

Add attachments to a Jira issue.

Parameters:

  • issue_id (str | int) –

    the Jira issue id

  • urls (list[str]) –

    list of urls to the attachments

Raises:

  • JiraAttachmentError –

    if there is an error while adding any attachment

Source code in src/firefighter/raid/client.py
@staticmethod
def add_attachments_to_issue(issue_id: str | int, urls: list[str]) -> None:
    """Add attachments to a Jira issue.

    Args:
        issue_id (str | int): the Jira issue id
        urls (list[str]): list of urls to the attachments

    Raises:
        JiraAttachmentError: if there is an error while adding any attachment
    """
    http_client = HttpClient()
    for i, url in enumerate(urls):
        index = url.rfind(".")
        extension = url[index:]
        try:
            response = http_client.get(url)
            response.raise_for_status()  # Raises an exception if status code is not 2XX

            in_memory_file = io.BytesIO(response.content)

            client.jira.add_attachment(
                issue=issue_id,
                attachment=in_memory_file,
                filename=f"image{i}{extension}",
            )

        except (HTTPError, JIRAError) as err:
            msg = f"Error while adding attachment to issue: {err}"
            raise JiraAttachmentError(msg) from err

assign_issue ¤

assign_issue(issue_key: str, account_id: str) -> bool

Assign a Jira issue to a user.

Parameters:

  • issue_key (str) –

    Jira issue key (e.g., "INCIDENT-123")

  • account_id (str) –

    Jira account ID of the user

Returns:

  • bool –

    True if assignment succeeded, False otherwise

Note

This method does not raise exceptions. Assignment failures are logged as warnings since assignment is typically an optional operation.

Source code in src/firefighter/jira_app/client.py
def assign_issue(self, issue_key: str, account_id: str) -> bool:
    """Assign a Jira issue to a user.

    Args:
        issue_key: Jira issue key (e.g., "INCIDENT-123")
        account_id: Jira account ID of the user

    Returns:
        True if assignment succeeded, False otherwise

    Note:
        This method does not raise exceptions. Assignment failures are logged
        as warnings since assignment is typically an optional operation.
    """
    try:
        self.jira.assign_issue(issue_key, account_id)
    except exceptions.JIRAError as e:
        logger.warning(
            "Failed to assign issue %s to user %s: %s",
            issue_key,
            account_id,
            e.text if hasattr(e, "text") else str(e),
        )
        return False
    else:
        logger.info("Assigned issue %s to user %s", issue_key, account_id)
        return True

create_postmortem_issue ¤

create_postmortem_issue(project_key: str, issue_type: str, fields: dict[str, Any], parent_issue_key: str | None = None) -> dict[str, Any]

Create a Jira post-mortem issue with custom fields.

Parameters:

  • project_key (str) –

    Jira project key (e.g., "INCIDENT")

  • issue_type (str) –

    Issue type name (e.g., "Post-mortem")

  • fields (dict[str, Any]) –

    Dictionary of field IDs to values

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

    Optional parent issue key to link this post-mortem to

Returns:

  • dict[str, Any] –

    Dictionary with 'key' and 'id' of created issue

Raises:

  • JiraAPIError –

    If issue creation fails

Source code in src/firefighter/jira_app/client.py
def create_postmortem_issue(
    self,
    project_key: str,
    issue_type: str,
    fields: dict[str, Any],
    parent_issue_key: str | None = None,
) -> dict[str, Any]:
    """Create a Jira post-mortem issue with custom fields.

    Args:
        project_key: Jira project key (e.g., "INCIDENT")
        issue_type: Issue type name (e.g., "Post-mortem")
        fields: Dictionary of field IDs to values
        parent_issue_key: Optional parent issue key to link this post-mortem to

    Returns:
        Dictionary with 'key' and 'id' of created issue

    Raises:
        JiraAPIError: If issue creation fails
    """
    try:
        issue_dict: dict[str, Any] = {
            "project": {"key": project_key},
            "issuetype": {"name": issue_type},
            **fields,
        }

        # Create the issue first without parent link
        issue = self.jira.create_issue(fields=issue_dict)
        logger.info(
            "Created post-mortem issue %s in project %s", issue.key, project_key
        )

        # Create issue link to parent if provided
        # Using link instead of parent to avoid hierarchy restrictions
        if parent_issue_key:
            self._create_issue_link_safe(
                parent_issue_key=parent_issue_key,
                postmortem_issue_key=issue.key,
            )

    except exceptions.JIRAError as e:
        logger.exception(
            "Failed to create Jira issue in project %s",
            project_key,
            extra={
                "ff.metric": "integration.error",
                "ff.integration": "jira",
                "ff.operation": "create_issue",
                "ff.jira_status_code": e.status_code,
            },
        )
        error_msg = f"Failed to create Jira issue: {e.status_code} {e.text}"
        raise JiraAPIError(error_msg) from e
    else:
        return {
            "key": issue.key,
            "id": issue.id,
        }

get_jira_user_from_jira_id ¤

get_jira_user_from_jira_id(jira_account_id: str) -> JiraUser

Look for a Jira User in DB, if not found, fetch it from Jira API.

Parameters:

  • jira_account_id (str) –

    Jira account id

Raises:

  • JiraUserNotFoundError –

    User not found in Jira nor in DB

  • JiraUserDatabaseError –

    Unable to create user in DB

  • ValueError –

    Empty jira_account_id

Returns:

  • JiraUser ( JiraUser ) –

    Jira user object

Source code in src/firefighter/jira_app/client.py
def get_jira_user_from_jira_id(self, jira_account_id: str) -> JiraUser:
    """Look for a Jira User in DB, if not found, fetch it from Jira API.

    Args:
        jira_account_id (str): Jira account id

    Raises:
        JiraUserNotFoundError: User not found in Jira nor in DB
        JiraUserDatabaseError: Unable to create user in DB
        ValueError: Empty jira_account_id

    Returns:
        JiraUser: Jira user object
    """
    if jira_account_id is None or jira_account_id == "":
        err_msg = f"Jira account id is empty ('{jira_account_id}')"
        raise ValueError(err_msg)

    # Look in the DB
    try:
        return JiraUser.objects.get(id=jira_account_id)
    except JiraUser.DoesNotExist:
        logger.debug(
            f"Jira user {jira_account_id} not found in DB, fetching from Jira API"
        )
    logger.info("User %s not found in DB. Check sync user task.", jira_account_id)

    # Look on JIRA API
    jira_api_user, email = self._get_user_from_api(jira_account_id)

    username: str = email.split("@")[0]
    # Check if we have user with same email
    try:
        user: User = User.objects.select_related("jira_user").get(email=email)
        if (
            hasattr(user, "jira_user")
            and user.jira_user
            and isinstance(user.jira_user, JiraUser)
        ):
            return user.jira_user
        try:
            return JiraUser.objects.create(
                id=jira_account_id,
                user=user,
            )
        except db.IntegrityError as e:
            logger.exception("Error creating user %s", jira_account_id)
            raise JiraUserDatabaseError("Unable to create user") from e

    except User.DoesNotExist:
        logger.warning("User %s not found in DB. Creating it...", jira_account_id)
        user = self._create_user_from_jira_info(
            jira_account_id, jira_api_user, email, username
        )

    try:
        return JiraUser.objects.create(
            id=jira_account_id,
            user=user,
        )
    except db.IntegrityError as e:
        logger.exception("Error creating user %s", jira_account_id)
        raise JiraUserDatabaseError("Unable to create user") from e

get_jira_user_from_user ¤

get_jira_user_from_user(user: User) -> JiraUser

Fetches a Jira user from the Jira API.

Parameters:

  • user (User) –

    User object

Raises:

  • JiraUserNotFoundError –

    User not found in Jira

Returns:

  • JiraAPIUser ( JiraUser ) –

    Jira API user object

Source code in src/firefighter/jira_app/client.py
def get_jira_user_from_user(self, user: User) -> JiraUser:
    """Fetches a Jira user from the Jira API.

    Args:
        user (User): User object

    Raises:
        JiraUserNotFoundError: User not found in Jira

    Returns:
        JiraAPIUser: Jira API user object
    """
    # Check if user has a Jira user
    if hasattr(user, "jira_user") and user.jira_user:
        return user.jira_user

    username = user.email.split("@")[0]
    jira_user = self._fetch_jira_user(username)

    return JiraUser.objects.update_or_create(
        id=jira_user.raw.get("accountId"), defaults={"user": user}
    )[0]

get_project_creation_problem ¤

get_project_creation_problem(project_key: str) -> str | None

Tell why issues cannot be created in a Jira project, or None if they can.

Catches projects archived or deleted in Jira, and projects our Jira account cannot see or create issues in.

Source code in src/firefighter/raid/client.py
def get_project_creation_problem(self, project_key: str) -> str | None:
    """Tell why issues cannot be created in a Jira project, or None if they can.

    Catches projects archived or deleted in Jira, and projects our Jira account
    cannot see or create issues in.
    """
    try:
        project = self.jira.project(project_key)
    except JIRAError as err:
        if err.status_code == 404:
            return "not found (deleted, or not visible to our Jira account)"
        raise
    if project.raw.get("archived"):
        return "archived"
    permissions = self.jira.my_permissions(
        projectKey=project_key, permissions="CREATE_ISSUES"
    )["permissions"]
    if not permissions["CREATE_ISSUES"]["havePermission"]:
        return "not open to issue creation for our Jira account"
    return None

get_watchers_from_jira_ticket ¤

get_watchers_from_jira_ticket(jira_issue_id: int | str) -> list[User]

Fetch watchers for a specific Jira ticket from Jira API.

Parameters:

  • jira_issue_id (str | int) –

    Jira issue id

Raises:

Returns:

  • list ( User ) –

    List of Jira users object, or empty list if ticket doesn't exist

Source code in src/firefighter/jira_app/client.py
def get_watchers_from_jira_ticket(
    self, jira_issue_id: int | str
) -> list[JiraAPIUser]:
    """Fetch watchers for a specific Jira ticket from Jira API.

    Args:
        jira_issue_id (str | int): Jira issue id

    Raises:
        ValueError: Empty issue id

    Returns:
        list(JiraAPIUser): List of Jira users object, or empty list if ticket doesn't exist
    """
    try:
        watchers = self.jira.watchers(jira_issue_id).raw.get("watchers")
    except exceptions.JIRAError as e:
        if e.status_code == 404:
            logger.warning(
                "Jira ticket %s not found or no permission to access it. Cannot fetch watchers.",
                jira_issue_id,
            )
            return []
        raise
    else:
        if len(watchers) == 0:
            logger.debug("No watchers found for jira_issue_id '%s'.", jira_issue_id)
        return watchers

transition_issue_auto ¤

transition_issue_auto(issue_id: str | int, target_status_name: str, workflow_name: str) -> None

Attempts to close an issue by applying transitions to it.

Parameters:

  • issue_id (str | int) –

    Jira issue id

  • target_status_name (str) –

    target status name

  • workflow_name (str) –

    workflow name

Source code in src/firefighter/jira_app/client.py
def transition_issue_auto(
    self, issue_id: str | int, target_status_name: str, workflow_name: str
) -> None:
    """Attempts to close an issue by applying transitions to it.

    Args:
        issue_id (str | int): Jira issue id
        target_status_name (str): target status name
        workflow_name (str): workflow name
    """
    issue_id = str(issue_id)
    transitions_info = self._get_transitions(
        self._get_project_config_workflow_from_builder_base(workflow_name)
    )
    if len(transitions_info) == 0:
        logger.error(
            f"Could not find transitions for issue id={issue_id}! Not closing issue."
        )

    # Get closed state id
    # XXX Use a list of closed states to support multiple workflows, or better
    closed_state_id = get_status_id_from_name(transitions_info, target_status_name)
    if closed_state_id is None:
        logger.warning(
            f"Could not find target status '{target_status_name}' id for issue {issue_id}! Not closing issue."
        )
        return

    # Get current issue status
    issue = self.jira.issue(issue_id)
    current_status_id = int(issue.fields.status.id)

    # Get transitions to apply
    transitions_to_apply = get_transitions_to_apply(
        current_status_id, transitions_info, closed_state_id
    )

    if len(transitions_to_apply) == 0:
        if current_status_id == closed_state_id:
            logger.info(
                f"Issue {issue_id} is already at target status "
                f"'{target_status_name}'. Nothing to do."
            )
        else:
            logger.warning(
                f"No transition path found for issue {issue_id} from status "
                f"id={current_status_id} to target '{target_status_name}'. "
                f"Leaving issue unchanged."
            )
        return

    # Apply transitions sequentially. A screen on a transition is fine as long
    # as it has no required field (we submit an empty field set). If a transition
    # ever fails (e.g. a field was later made required), stop and surface the
    # error rather than leaving the issue half-transitioned.
    for transition in transitions_to_apply:
        logger.debug(f"Running transition: {transition}")
        try:
            self.jira.transition_issue(
                issue=issue_id,
                transition=transition,
                fields={},
            )
        except Exception:
            logger.exception(
                "Failed to apply transition '%s' on issue %s while moving to "
                "'%s'. Aborting remaining transitions; issue may be partially "
                "transitioned.",
                transition,
                issue_id,
                target_status_name,
            )
            raise

update_issue_fields ¤

update_issue_fields(issue_id: str | int, **fields: Any) -> None

Update Jira issue fields (supports custom fields like customfield_11064).

Source code in src/firefighter/raid/client.py
def update_issue_fields(self, issue_id: str | int, **fields: Any) -> None:
    """Update Jira issue fields (supports custom fields like customfield_11064)."""
    issue = self.jira.issue(str(issue_id))
    issue.update(fields=fields)

forms ¤

Functions:

prepare_jira_fields ¤

prepare_jira_fields(
    *,
    title: str,
    description: str,
    priority: int,
    reporter: str,
    incident_category: str,
    environments: list[str],
    platforms: list[str],
    impacts_data: dict[str, ImpactLevel],
    optional_fields: dict[str, Any] | None = None
) -> dict[str, Any]

Prepare all fields for jira_client.create_issue().

This function centralizes Jira field preparation for both P1-P3 and P4-P5 incidents, ensuring all custom fields are properly passed.

Parameters:

  • title (str) –

    Incident title

  • description (str) –

    Incident description

  • priority (int) –

    Priority value (1-5)

  • reporter (str) –

    Jira user account ID

  • incident_category (str) –

    Category name

  • environments (list[str]) –

    List of environment values (e.g. ["PRD", "STG"])

  • platforms (list[str]) –

    List of platform values (e.g. ["platform-FR", "platform-DE"])

  • impacts_data (dict[str, ImpactLevel]) –

    Dictionary of impact data for business_impact computation

  • optional_fields (dict[str, Any] | None, default: None ) –

    Optional dictionary containing: - zendesk_ticket_id: Zendesk ticket ID (customer-specific) - seller_contract_id: Seller contract ID (seller-specific) - zoho_desk_ticket_id: Zoho Desk ticket ID (seller-specific) - is_key_account: Key account flag (seller-specific) - is_seller_in_golden_list: Golden list flag (seller-specific) - suggested_team_routing: Suggested team routing (P4-P5 only)

Returns:

  • dict[str, Any] –

    Dictionary of kwargs ready for jira_client.create_issue()

Source code in src/firefighter/raid/forms.py
def prepare_jira_fields(
    *,
    title: str,
    description: str,
    priority: int,
    reporter: str,
    incident_category: str,
    environments: list[str],
    platforms: list[str],
    impacts_data: dict[str, ImpactLevel],
    optional_fields: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Prepare all fields for jira_client.create_issue().

    This function centralizes Jira field preparation for both P1-P3 and P4-P5 incidents,
    ensuring all custom fields are properly passed.

    Args:
        title: Incident title
        description: Incident description
        priority: Priority value (1-5)
        reporter: Jira user account ID
        incident_category: Category name
        environments: List of environment values (e.g. ["PRD", "STG"])
        platforms: List of platform values (e.g. ["platform-FR", "platform-DE"])
        impacts_data: Dictionary of impact data for business_impact computation
        optional_fields: Optional dictionary containing:
            - zendesk_ticket_id: Zendesk ticket ID (customer-specific)
            - seller_contract_id: Seller contract ID (seller-specific)
            - zoho_desk_ticket_id: Zoho Desk ticket ID (seller-specific)
            - is_key_account: Key account flag (seller-specific)
            - is_seller_in_golden_list: Golden list flag (seller-specific)
            - suggested_team_routing: Suggested team routing (P4-P5 only)

    Returns:
        Dictionary of kwargs ready for jira_client.create_issue()
    """
    business_impact = get_business_impact(impacts_data)
    platform = platforms[0] if platforms else PlatformChoices.ALL.value

    # Extract optional fields with defaults
    opt = optional_fields or {}
    zendesk_ticket_id = opt.get("zendesk_ticket_id", "")
    seller_contract_id = opt.get("seller_contract_id", "")
    zoho_desk_ticket_id = opt.get("zoho_desk_ticket_id", "")
    is_key_account = opt.get("is_key_account")
    is_seller_in_golden_list = opt.get("is_seller_in_golden_list")
    suggested_team_routing = opt.get("suggested_team_routing")

    return {
        "issuetype": "Incident",
        "summary": title,
        "description": description,
        "priority": priority,
        "reporter": reporter,
        "assignee": None,
        "incident_category": incident_category,
        "environments": environments,  # ✅ Always pass environments list
        "platform": platform,
        "business_impact": business_impact,
        "zendesk_ticket_id": zendesk_ticket_id,
        "seller_contract_id": seller_contract_id,
        "zoho_desk_ticket_id": zoho_desk_ticket_id,
        "is_key_account": is_key_account if is_key_account is not None else False,
        "is_seller_in_golden_list": is_seller_in_golden_list if is_seller_in_golden_list is not None else False,
        "suggested_team_routing": suggested_team_routing,
    }

messages ¤

Classes:

SlackMessageJiraClosedBlockedByCanBeClosed ¤

SlackMessageJiraClosedBlockedByCanBeClosed(jira_ticket_key: str, reasons: list[tuple[str, str]])

Bases: SlackMessageSurface

Sent to the incident channel when a Jira 'Closed' webhook was received but the incident cannot be automatically closed (e.g. missing key events).

Methods:

  • get_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

  • post_message –

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

Attributes:

Source code in src/firefighter/raid/messages.py
def __init__(self, jira_ticket_key: str, reasons: list[tuple[str, str]]) -> None:
    self.jira_ticket_key = jira_ticket_key
    self.reasons = reasons
    super().__init__()

strategy class-attribute instance-attribute ¤

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(),
    )

migrations ¤

Modules:

0004_add_check_feature_team_projects_periodic_task ¤

Functions:

  • create_periodic_task –

    Check every Monday at 9:00 (Europe/Paris) that feature team Jira projects accept tickets.

create_periodic_task ¤

create_periodic_task(apps, schema_editor)

Check every Monday at 9:00 (Europe/Paris) that feature team Jira projects accept tickets.

Source code in src/firefighter/raid/migrations/0004_add_check_feature_team_projects_periodic_task.py
def create_periodic_task(apps, schema_editor):
    """Check every Monday at 9:00 (Europe/Paris) that feature team Jira projects accept tickets."""
    CrontabSchedule = apps.get_model("django_celery_beat", "CrontabSchedule")
    PeriodicTask = apps.get_model("django_celery_beat", "PeriodicTask")

    schedule, _ = CrontabSchedule.objects.get_or_create(
        minute="0",
        hour="9",
        day_of_week="1",
        day_of_month="*",
        month_of_year="*",
        timezone="Europe/Paris",
    )
    PeriodicTask.objects.get_or_create(
        name="Check feature team Jira projects",
        defaults={
            "task": TASK_NAME,
            "crontab": schedule,
            "enabled": True,
            "description": "Log an error for each feature team whose Jira project is archived, deleted or closed to issue creation (Mondays at 9 AM)",
        },
    )

models ¤

Classes:

JiraTicket ¤

Bases: JiraIssue

Jira ticket model.

serializers ¤

Classes:

Functions:

JiraWebhookUpdateSerializer ¤

Bases: Serializer[Any]

LandbotIssueRequestSerializer ¤

Bases: ModelSerializer[JiraTicket]

Methods:

validate_labels ¤

validate_labels(value: list[str] | None) -> list[str]

Transform null labels to empty list.

Source code in src/firefighter/raid/serializers.py
def validate_labels(self, value: list[str] | None) -> list[str]:
    """Transform null labels to empty list."""
    if value is None:
        return []
    return value

parse_attachment_urls ¤

parse_attachment_urls(raw: str | None) -> list[str]

Normalise the attachments payload sent by Landbot into a list of URLs.

Landbot historically sends a Python-stringified list (e.g. "['https://a', 'https://b']") rather than a JSON array. This helper tolerates that legacy format along with a plain comma-separated string or a single URL.

Source code in src/firefighter/raid/serializers.py
def parse_attachment_urls(raw: str | None) -> list[str]:
    """Normalise the attachments payload sent by Landbot into a list of URLs.

    Landbot historically sends a Python-stringified list (e.g. ``"['https://a', 'https://b']"``)
    rather than a JSON array. This helper tolerates that legacy format along with
    a plain comma-separated string or a single URL.
    """
    if not raw:
        return []
    stripped = raw.replace("[", "").replace("]", "").replace("'", "").replace('"', "")
    return [item.strip() for item in stripped.split(",") if item.strip()]

validate_no_spaces ¤

validate_no_spaces(value: str) -> None

Ensure the string does not contain spaces.

Source code in src/firefighter/raid/serializers.py
def validate_no_spaces(value: str) -> None:
    """Ensure the string does not contain spaces."""
    if " " in value:
        raise serializers.ValidationError("The string cannot contain spaces.")

service ¤

Classes:

Functions:

CustomerIssueData dataclass ¤

CustomerIssueData(
    priority: int | None,
    labels: list[str] | None,
    platform: str,
    business_impact: str | None,
    team_to_be_routed: str | None,
    area: str | None,
    zendesk_ticket_id: str | None,
    incident_category: str | None = None,
)

Data container for customer issue creation parameters.

check_issue_id ¤

check_issue_id(issue: JiraObject, title: str, reporter: str) -> int | str

Check and return the issue ID from a JiraObject.

Raises JiraAPIError if the issue ID is missing or invalid.

Source code in src/firefighter/raid/service.py
def check_issue_id(issue: JiraObject, title: str, reporter: str) -> int | str:
    """Check and return the issue ID from a JiraObject.

    Raises JiraAPIError if the issue ID is missing or invalid.
    """
    issue_id = issue.get("id")
    if issue_id is None:
        logger.error(
            f"Could not create Jira ticket for the incident {title} and the reporter {reporter}"
        )
        raise JiraAPIError(error_jira_ticket_creation)
    return issue_id

create_issue_customer ¤

create_issue_customer(title: str, description: str, reporter: str, issue_data: CustomerIssueData) -> JiraObject

Creates a Jira Incident issue of type Customer.

Parameters:

  • title (str) –

    Summary of the issue

  • description (str) –

    Description of the issue

  • reporter (str) –

    Jira account id of the reporter

  • issue_data (CustomerIssueData) –

    Container with issue parameters

Source code in src/firefighter/raid/service.py
def create_issue_customer(
    title: str,
    description: str,
    reporter: str,
    issue_data: CustomerIssueData,
) -> JiraObject:
    """Creates a Jira Incident issue of type Customer.

    Args:
        title (str): Summary of the issue
        description (str): Description of the issue
        reporter (str): Jira account id of the reporter
        issue_data (CustomerIssueData): Container with issue parameters
    """
    issue = jira_client.create_issue(
        issuetype="Incident",
        summary=title,
        description=description,
        assignee=None,
        reporter=reporter,
        priority=issue_data.priority,
        labels=issue_data.labels,
        platform=issue_data.platform,
        business_impact=issue_data.business_impact,
        suggested_team_routing=issue_data.team_to_be_routed,
        zendesk_ticket_id=issue_data.zendesk_ticket_id,
        incident_category=issue_data.incident_category,
    )
    check_issue_id(issue, title=title, reporter=reporter)
    return issue

create_issue_documentation_request ¤

create_issue_documentation_request(
    title: str, description: str, reporter: str, priority: int | None, labels: list[str] | None, platform: str
) -> JiraObject

Creates a Jira issue of type Documentation/Process Request.

Parameters:

  • title (str) –

    Summary of the issue

  • description (str) –

    Description of the issue

  • reporter (str) –

    Jira account id of the reporter

  • priority (int) –

    Priority of the issue

  • labels (list[str]) –

    Labels to add to the issue

  • platform (str) –

    Platform of the issue

Source code in src/firefighter/raid/service.py
def create_issue_documentation_request(
    title: str,
    description: str,
    reporter: str,
    priority: int | None,
    labels: list[str] | None,
    platform: str,
) -> JiraObject:
    """Creates a Jira issue of type Documentation/Process Request.

    Args:
        title (str): Summary of the issue
        description (str): Description of the issue
        reporter (str): Jira account id of the reporter
        priority (int): Priority of the issue
        labels (list[str]): Labels to add to the issue
        platform (str): Platform of the issue
    """
    if labels is None:
        labels = [""]
    if "documentation-request" not in labels:
        labels.append("documentation-request")
    issue = jira_client.create_issue(
        issuetype="Documentation/Process Request",
        summary=title,
        description=description,
        assignee=None,
        reporter=reporter,
        priority=priority,
        labels=labels,
        platform=platform,
    )
    check_issue_id(issue, title=title, reporter=reporter)
    return issue

create_issue_feature_request ¤

create_issue_feature_request(
    title: str, description: str, reporter: str, priority: int | None, labels: list[str] | None, platform: str
) -> JiraObject

Creates a Jira issue of type Feature Request.

Parameters:

  • title (str) –

    Summary of the issue

  • description (str) –

    Description of the issue

  • reporter (str) –

    Jira account id of the reporter

  • priority (int) –

    Priority of the issue

  • labels (list[str]) –

    Labels to add to the issue

  • platform (str) –

    Platform of the issue

Source code in src/firefighter/raid/service.py
def create_issue_feature_request(
    title: str,
    description: str,
    reporter: str,
    priority: int | None,
    labels: list[str] | None,
    platform: str,
) -> JiraObject:
    """Creates a Jira issue of type Feature Request.

    Args:
        title (str): Summary of the issue
        description (str): Description of the issue
        reporter (str): Jira account id of the reporter
        priority (int): Priority of the issue
        labels (list[str]): Labels to add to the issue
        platform (str): Platform of the issue
    """
    if labels is None:
        labels = [""]
    if "feature-request" not in labels:
        labels.append("feature-request")
    issue = jira_client.create_issue(
        issuetype="Feature Request",
        summary=title,
        description=description,
        assignee=None,
        reporter=reporter,
        priority=priority,
        labels=labels,
        platform=platform,
    )
    check_issue_id(issue, title=title, reporter=reporter)
    return issue

create_issue_internal ¤

create_issue_internal(
    title: str,
    description: str,
    reporter: str,
    priority: int | None,
    labels: list[str] | None,
    platform: str,
    business_impact: str | None,
    team_to_be_routed: str | None,
    incident_category: str | None,
) -> JiraObject

Creates a Jira Incident Issue of type Internal.

Parameters:

  • title (str) –

    Summary of the issue

  • description (str) –

    Description of the issue

  • reporter (str) –

    Jira account id of the reporter

  • priority (int) –

    Priority of the issue

  • labels (list[str]) –

    Labels to add to the issue

  • platform (str) –

    Platform of the issue

  • business_impact (str) –

    Business impact of the issue

  • team_to_be_routed (str) –

    Team to be routed

  • incident_category (str) –

    Incident category of the issue

Source code in src/firefighter/raid/service.py
def create_issue_internal(
    title: str,
    description: str,
    reporter: str,
    priority: int | None,
    labels: list[str] | None,
    platform: str,
    business_impact: str | None,
    team_to_be_routed: str | None,
    incident_category: str | None,
) -> JiraObject:
    """Creates a Jira Incident Issue of type Internal.

    Args:
        title (str): Summary of the issue
        description (str): Description of the issue
        reporter (str): Jira account id of the reporter
        priority (int): Priority of the issue
        labels (list[str]): Labels to add to the issue
        platform (str): Platform of the issue
        business_impact (str): Business impact of the issue
        team_to_be_routed (str): Team to be routed
        incident_category (str): Incident category of the issue
    """
    issue = jira_client.create_issue(
        issuetype="Incident",
        summary=title,
        description=description,
        assignee=None,
        reporter=reporter,
        priority=priority,
        labels=labels,
        platform=platform,
        business_impact=business_impact,
        suggested_team_routing=team_to_be_routed,
        incident_category=incident_category,
    )
    check_issue_id(issue, title=title, reporter=reporter)
    return issue

create_issue_seller ¤

create_issue_seller(
    *,
    title: str,
    description: str,
    reporter: str,
    priority: int | None,
    labels: list[str] | None,
    platform: str,
    business_impact: str | None,
    team_to_be_routed: str | None,
    incident_category: str | None,
    seller_contract_id: str | None,
    is_key_account: bool | None,
    is_seller_in_golden_list: bool | None,
    zoho_desk_ticket_id: str | None
) -> JiraObject

Creates a Jira Incident issue of type Seller.

Parameters:

  • title (str) –

    Summary of the issue

  • description (str) –

    Description of the issue

  • reporter (str) –

    Jira account id of the reporter

  • priority (int) –

    Priority of the issue

  • labels (list[str]) –

    Labels to add to the issue

  • platform (str) –

    Platform of the issue

  • business_impact (str) –

    Business impact of the issue

  • team_to_be_routed (str) –

    Team to be routed

  • incident_category (str) –

    Incident category of the issue

  • seller_contract_id (str) –

    Seller contract id

  • is_key_account (bool) –

    Is key account

  • is_seller_in_golden_list (bool) –

    Is seller in golden list

  • zoho_desk_ticket_id (str) –

    Zoho desk ticket id

Source code in src/firefighter/raid/service.py
def create_issue_seller(  # noqa: PLR0913
    *,
    title: str,
    description: str,
    reporter: str,
    priority: int | None,
    labels: list[str] | None,
    platform: str,
    business_impact: str | None,
    team_to_be_routed: str | None,
    incident_category: str | None,
    seller_contract_id: str | None,
    is_key_account: bool | None,
    is_seller_in_golden_list: bool | None,
    zoho_desk_ticket_id: str | None,
) -> JiraObject:
    """Creates a Jira Incident issue of type Seller.

    Args:
        title (str): Summary of the issue
        description (str): Description of the issue
        reporter (str): Jira account id of the reporter
        priority (int): Priority of the issue
        labels (list[str]): Labels to add to the issue
        platform (str): Platform of the issue
        business_impact (str): Business impact of the issue
        team_to_be_routed (str): Team to be routed
        incident_category (str): Incident category of the issue
        seller_contract_id (str): Seller contract id
        is_key_account (bool): Is key account
        is_seller_in_golden_list (bool): Is seller in golden list
        zoho_desk_ticket_id (str): Zoho desk ticket id
    """
    issue = jira_client.create_issue(
        issuetype="Incident",
        summary=title,
        description=description,
        assignee=None,
        reporter=reporter,
        priority=priority,
        labels=labels,
        platform=platform,
        business_impact=business_impact,
        suggested_team_routing=team_to_be_routed,
        seller_contract_id=seller_contract_id,
        is_key_account=is_key_account,
        is_seller_in_golden_list=is_seller_in_golden_list,
        zoho_desk_ticket_id=zoho_desk_ticket_id,
        incident_category=incident_category,
    )
    check_issue_id(issue, title=title, reporter=reporter)
    return issue

get_jira_user_from_user ¤

get_jira_user_from_user(user: User) -> JiraUser

Returns the JiraUser object for a given user, if it exists in DB or can be fetched on Jira API. Returns the default JiraUser if not found.

Source code in src/firefighter/raid/service.py
def get_jira_user_from_user(user: User) -> JiraUser:
    """Returns the JiraUser object for a given user, if it exists in DB or can be fetched on Jira API. Returns the default JiraUser if not found."""
    try:
        jira_user = jira_client.get_jira_user_from_user(user)
    except JiraAPIError:
        logger.exception(f"Could not find Jira user for {user.id}")

        logger.warning(f"User {user.id} has no Jira user")
        try:
            jira_user = jira_client.get_jira_user_from_jira_id(
                RAID_DEFAULT_JIRA_QRAFT_USER_ID
            )
        except JiraUserNotFoundError:
            logger.exception(
                f"Could not find Jira user with account id {RAID_DEFAULT_JIRA_QRAFT_USER_ID}"
            )
            jira_user = JiraUser.objects.get(id=RAID_DEFAULT_JIRA_QRAFT_USER_ID)
    return jira_user

signals ¤

Modules:

incident_updated ¤

Functions:

incident_priority_post_save_fallback ¤

incident_priority_post_save_fallback(
    sender: Any, instance: Incident, *, created: bool, update_fields: set[str] | None, **kwargs: Any
) -> None

Fallback to push priority to Jira when Incident saves with priority_id in update_fields but no incident_updated signal fired (e.g., admin edits). Skips when marked to avoid loops.

Source code in src/firefighter/raid/signals/incident_updated.py
@receiver(post_save, sender=Incident)
def incident_priority_post_save_fallback(
    sender: Any,
    instance: Incident,
    *,
    created: bool,
    update_fields: set[str] | None,
    **kwargs: Any,
) -> None:
    """Fallback to push priority to Jira when Incident saves with priority_id in update_fields
    but no incident_updated signal fired (e.g., admin edits). Skips when marked to avoid loops.
    """
    if created:
        return
    if update_fields and "priority_id" not in update_fields:
        return
    if getattr(instance, "_skip_priority_sync", False):
        logger.debug(
            "Skipping post_save priority sync for incident #%s due to skip flag",
            getattr(instance, "id", "unknown"),
        )
        return
    if not hasattr(instance, "jira_ticket") or instance.jira_ticket is None:
        logger.debug(
            "Skipping post_save priority sync: incident #%s has no Jira ticket",
            getattr(instance, "id", "unknown"),
        )
        return
    if not instance.priority:
        logger.debug(
            "Skipping post_save priority sync: incident #%s priority missing",
            getattr(instance, "id", "unknown"),
        )
        return

    try:
        _set_impact_to_jira_cache(instance.id, "priority", instance.priority.value)
        client.update_issue_fields(
            instance.jira_ticket.id,
            customfield_11064={"value": str(instance.priority.value)},
        )
        logger.info(
            "Post-save synced priority %s to Jira ticket %s (customfield_11064) for incident #%s",
            instance.priority.value,
            instance.jira_ticket.id,
            getattr(instance, "id", "unknown"),
        )
    except Exception:
        logger.exception(
            "Failed post-save priority sync %s to Jira ticket %s for incident %s",
            instance.priority.value,
            instance.jira_ticket.id,
            getattr(instance, "id", "unknown"),
        )

incident_status_post_save_fallback ¤

incident_status_post_save_fallback(
    sender: Any, instance: Incident, *, created: bool, update_fields: set[str] | None, **kwargs: Any
) -> None

Fallback to push status to Jira when Incident saves with status in update_fields but no incident_updated signal fired (e.g., admin edits). Skips when marked to avoid loops.

Source code in src/firefighter/raid/signals/incident_updated.py
@receiver(post_save, sender=Incident)
def incident_status_post_save_fallback(
    sender: Any,
    instance: Incident,
    *,
    created: bool,
    update_fields: set[str] | None,
    **kwargs: Any,
) -> None:
    """Fallback to push status to Jira when Incident saves with status in update_fields
    but no incident_updated signal fired (e.g., admin edits). Skips when marked to avoid loops.
    """
    if created:
        return
    if (
        update_fields
        and "_status" not in update_fields
        and "status" not in update_fields
    ):
        return
    if getattr(instance, "_skip_status_sync", False):
        logger.debug(
            "Skipping post_save status sync for incident #%s due to skip flag",
            getattr(instance, "id", "unknown"),
        )
        return
    if not hasattr(instance, "jira_ticket") or instance.jira_ticket is None:
        logger.debug(
            "Skipping post_save status sync: incident #%s has no Jira ticket",
            getattr(instance, "id", "unknown"),
        )
        return
    target_jira_status = IMPACT_TO_JIRA_STATUS_MAP.get(instance.status)
    if target_jira_status is None:
        logger.debug(
            "Skipping post_save status sync: no Jira mapping for status %s (incident #%s)",
            instance.status,
            getattr(instance, "id", "unknown"),
        )
        return
    try:
        _set_impact_to_jira_cache(instance.id, "status", target_jira_status)
        client.transition_issue_auto(
            instance.jira_ticket.id, target_jira_status, RAID_JIRA_WORKFLOW_NAME
        )
        logger.info(
            "Post-save synced status %s to Jira ticket %s for incident #%s",
            instance.status,
            instance.jira_ticket.id,
            getattr(instance, "id", "unknown"),
        )
    except Exception:
        logger.exception(
            "Failed post-save status sync %s to Jira ticket %s for incident %s",
            instance.status,
            instance.jira_ticket.id,
            getattr(instance, "id", "unknown"),
        )

incident_updated_close_ticket_when_mitigated_or_postmortem ¤

incident_updated_close_ticket_when_mitigated_or_postmortem(
    sender: Any, incident: Incident, incident_update: IncidentUpdate, updated_fields: list[str], **kwargs: Any
) -> None

Close Jira incident ticket based on incident status and priority.

Closure logic: - P1/P2 (needs_postmortem): Close only when incident is CLOSED - P3+ (no postmortem): Close when incident is MITIGATED or CLOSED - POST_MORTEM status never closes the ticket (it remains open during PM phase)

Source code in src/firefighter/raid/signals/incident_updated.py
@receiver(signal=incident_updated, sender="update_status")
def incident_updated_close_ticket_when_mitigated_or_postmortem(
    sender: Any,
    incident: Incident,
    incident_update: IncidentUpdate,
    updated_fields: list[str],
    **kwargs: Any,
) -> None:
    """Close Jira incident ticket based on incident status and priority.

    Closure logic:
    - P1/P2 (needs_postmortem): Close only when incident is CLOSED
    - P3+ (no postmortem): Close when incident is MITIGATED or CLOSED
    - POST_MORTEM status never closes the ticket (it remains open during PM phase)
    """
    logger.debug(
        "incident_updated handler invoked for incident #%s with status %s; updated_fields=%s event_type=%s",
        getattr(incident, "id", "unknown"),
        incident_update.status,
        updated_fields,
        incident_update.event_type,
    )
    # Skip if this update was produced by Jira webhook sync to avoid redundant close calls
    if incident_update.event_type == "jira_status_sync":
        logger.debug(
            "Skipping Jira transition: incident #%s update came from Jira (event_type=jira_status_sync)",
            getattr(incident, "id", "unknown"),
        )
        return

    if "_status" not in updated_fields:
        logger.debug(
            "Skipping Jira transition: incident #%s update lacks _status in updated_fields (%s)",
            getattr(incident, "id", "unknown"),
            updated_fields,
        )
        return

    if not hasattr(incident, "jira_ticket") or incident.jira_ticket is None:
        logger.warning(
            "Trying to close Jira ticket for incident %s but no Jira ticket found",
            getattr(incident, "id", "unknown"),
        )
        return

    # Special case: when Impact moves to INVESTIGATING or MITIGATING, Jira must go
    # through two steps: "Pending resolution" then "in progress". We also set
    # cache keys for both steps to prevent the Jira→Impact webhook from bouncing
    # Impact back to OPEN on the intermediate hop.
    if incident_update.status in (IncidentStatus.INVESTIGATING, IncidentStatus.MITIGATING):
        steps = (JIRA_STATUS_PENDING_RESOLUTION, JIRA_STATUS_IN_PROGRESS)
        incident_id = getattr(incident, "id", None)
        if incident_id is not None:
            for step in steps:
                _set_impact_to_jira_cache(incident_id, "status", step)

        all_steps_succeeded = True
        for step in steps:
            try:
                logger.debug(
                    "Transitioning Jira ticket %s via workflow %s to status %s (incident #%s, impact status %s)",
                    incident.jira_ticket.id,
                    RAID_JIRA_WORKFLOW_NAME,
                    step,
                    getattr(incident, "id", "unknown"),
                    incident_update.status,
                )
                client.transition_issue_auto(
                    incident.jira_ticket.id, step, RAID_JIRA_WORKFLOW_NAME
                )
            except Exception:
                all_steps_succeeded = False
                logger.exception(
                    "Failed to transition Jira ticket %s to %s for incident %s",
                    incident.jira_ticket.id,
                    step,
                    getattr(incident, "id", "unknown"),
                )
        if all_steps_succeeded:
            logger.info(
                "Transitioned Jira ticket %s through Pending resolution -> in progress from Impact status %s",
                incident.jira_ticket.id,
                incident_update.status.label if incident_update.status else "Unknown",
            )
        else:
            logger.warning(
                "At least one Jira transition failed while moving ticket %s to MITIGATING (incident #%s)",
                incident.jira_ticket.id,
                getattr(incident, "id", "unknown"),
            )
        return

    # Decide target Jira status based on Impact status and postmortem requirement.
    # P3+ (no postmortem): close Jira when Impact reaches MITIGATED or CLOSED.
    # P1/P2 (needs_postmortem): close Jira only when Impact reaches CLOSED.
    incident_status = incident_update.status
    if incident_status is None:
        logger.info(
            "Skipping Jira transition: incident #%s status is None",
            getattr(incident, "id", "unknown"),
        )
        return

    target_jira_status: str | None = (
        JIRA_STATUS_CLOSED
        if incident_status == IncidentStatus.CLOSED
        else IMPACT_TO_JIRA_STATUS_MAP.get(incident_status)
    )

    if target_jira_status is None:
        logger.info(
            "Skipping Jira transition: no Jira status mapping for Impact status %s (incident #%s)",
            incident_update.status,
            getattr(incident, "id", "unknown"),
        )
        return

    try:
        incident_id = getattr(incident, "id", None)
        if incident_id is not None:
            _set_impact_to_jira_cache(incident_id, "status", target_jira_status)
        logger.debug(
            "Transitioning Jira ticket %s via workflow %s to status %s (incident #%s, impact status %s)",
            incident.jira_ticket.id,
            RAID_JIRA_WORKFLOW_NAME,
            target_jira_status,
            getattr(incident, "id", "unknown"),
            incident_update.status,
        )
        client.transition_issue_auto(
            incident.jira_ticket.id, target_jira_status, RAID_JIRA_WORKFLOW_NAME
        )
        logger.info(
            "Transitioned Jira ticket %s to %s from Impact status %s",
            incident.jira_ticket.id,
            target_jira_status,
            incident_update.status.label if incident_update.status else "Unknown",
        )
    except Exception:
        logger.exception(
            "Failed to transition Jira ticket %s to %s for incident %s",
            incident.jira_ticket.id,
            target_jira_status,
            getattr(incident, "id", "unknown"),
        )

incident_updated_sync_priority_to_jira ¤

incident_updated_sync_priority_to_jira(
    sender: Any, incident: Incident, incident_update: IncidentUpdate, updated_fields: list[str], **kwargs: Any
) -> None

Push Impact priority changes to Jira custom priority field (customfield_11064). Skips if change originated from Jira (event_type='jira_priority_sync') to avoid loops.

Source code in src/firefighter/raid/signals/incident_updated.py
@receiver(signal=incident_updated)
def incident_updated_sync_priority_to_jira(
    sender: Any,
    incident: Incident,
    incident_update: IncidentUpdate,
    updated_fields: list[str],
    **kwargs: Any,
) -> None:
    """Push Impact priority changes to Jira custom priority field (customfield_11064).
    Skips if change originated from Jira (event_type='jira_priority_sync') to avoid loops.
    """
    logger.debug(
        "Priority sync handler invoked: incident #%s sender=%s updated_fields=%s event_type=%s",
        getattr(incident, "id", "unknown"),
        sender,
        updated_fields,
        incident_update.event_type,
    )

    if incident_update.event_type == "jira_priority_sync":
        logger.debug(
            "Skipping Jira priority sync: incident #%s update came from Jira (event_type=jira_priority_sync)",
            getattr(incident, "id", "unknown"),
        )
        return

    if "priority_id" not in updated_fields:
        logger.debug(
            "Skipping Jira priority sync: incident #%s update lacks priority_id in updated_fields (%s) sender=%s",
            getattr(incident, "id", "unknown"),
            updated_fields,
            sender,
        )
        return

    if not hasattr(incident, "jira_ticket") or incident.jira_ticket is None:
        logger.debug(
            "Skipping Jira priority sync: incident #%s has no Jira ticket",
            getattr(incident, "id", "unknown"),
        )
        return

    if not incident.priority:
        logger.debug(
            "Skipping Jira priority sync: incident #%s priority is missing",
            getattr(incident, "id", "unknown"),
        )
        return

    try:
        incident_id = getattr(incident, "id", None)
        if incident_id is not None:
            _set_impact_to_jira_cache(incident_id, "priority", incident.priority.value)
        client.update_issue_fields(
            incident.jira_ticket.id,
            customfield_11064={"value": str(incident.priority.value)},
        )
        logger.info(
            "Synced priority %s to Jira ticket %s (customfield_11064) for incident #%s",
            incident.priority.value,
            incident.jira_ticket.id,
            getattr(incident, "id", "unknown"),
        )
    except Exception:
        logger.exception(
            "Failed to sync priority %s to Jira ticket %s for incident %s",
            incident.priority.value,
            incident.jira_ticket.id,
            getattr(incident, "id", "unknown"),
        )

tasks ¤

RAID Celery tasks.

Modules:

check_feature_team_projects ¤

Functions:

check_feature_team_projects ¤

check_feature_team_projects() -> None

Log an error for each feature team whose Jira project cannot receive tickets.

Jira projects get archived or restricted without Impact being told: the tickets of such a team silently fall back to the default project.

Source code in src/firefighter/raid/tasks/check_feature_team_projects.py
@shared_task(name="raid.check_feature_team_projects")
def check_feature_team_projects() -> None:
    """Log an error for each feature team whose Jira project cannot receive tickets.

    Jira projects get archived or restricted without Impact being told: the tickets
    of such a team silently fall back to the default project.
    """
    for team in FeatureTeam.objects.order_by("jira_project_key"):
        key = team.jira_project_key
        try:
            problem = client.get_project_creation_problem(key)
        except JIRAError:
            logger.exception(f"Could not check the Jira project {key} of feature team {team.name}")
            continue
        if problem:
            logger.error(
                f"Feature team {team.name} routes to Jira project {key}, which is {problem}: "
                f"its tickets fall back to {RAID_JIRA_PROJECT_KEY}. "
                "Update or remove this feature team."
            )

utils ¤

Functions:

get_domain_from_email cached ¤

get_domain_from_email(email: str) -> str

Returns the domain from an email address.

Removes any subdomain(s) and the @, applies .lower().

Parameters:

  • email (str) –

    The email address to extract the domain from.

Returns:

  • str –

    The domain part of the email address.

Raises:

Examples:

  • john.doe@example.com => example.com
  • alice.bob@test.example.ORG => example.org
  • webmaster@localhost => localhost
Source code in src/firefighter/raid/utils.py
@cache
def get_domain_from_email(email: str) -> str:
    """Returns the domain from an email address.

    Removes any subdomain(s) and the @, applies .lower().

    Args:
        email: The email address to extract the domain from.

    Returns:
        The domain part of the email address.

    Raises:
        ValueError: If the email is not well-formed.

    Examples:
      - `john.doe@example.com` => `example.com`
      - `alice.bob@test.example.ORG` => `example.org`
      - `webmaster@localhost` => `localhost`
    """
    # If there is not exactly one @, the email is invalid
    if email.count("@") != 1:
        msg = f"Invalid email: {email}"
        raise ValueError(msg)
    domain = email.rsplit("@", maxsplit=1)[-1]
    if not domain:
        msg = f"Invalid email: {email}"
        raise ValueError(msg)
    domain_parts = domain.split(".")

    return (".".join(domain_parts[-2:]) if len(domain_parts) > 2 else domain).lower()

normalize_cache_value ¤

normalize_cache_value(value: Any) -> str

Normalize cache values for loop-prevention keys.

Source code in src/firefighter/raid/utils.py
def normalize_cache_value(value: Any) -> str:
    """Normalize cache values for loop-prevention keys."""
    if value is None:
        return ""
    if isinstance(value, str):
        return value.strip().lower()
    try:
        return str(int(value))
    except (TypeError, ValueError):
        return str(value).strip().lower()

views ¤

Classes:

CreateJiraBotView ¤

Bases: CreateModelMixin, GenericAPIView[JiraTicket]

Methods:

  • post –

    Allow to create a Jira ticket through Landbot.

post ¤

post(request: Request, *args: Never, **kwargs: Never) -> Response

Allow to create a Jira ticket through Landbot. Requires a valid Bearer token, that you can create in the back-office if you have the right permissions.

Source code in src/firefighter/raid/views/__init__.py
def post(self, request: Request, *args: Never, **kwargs: Never) -> Response:
    """Allow to create a Jira ticket through Landbot.
    Requires a valid Bearer token, that you can create in the back-office if you have the right permissions.
    """
    serializer = self.get_serializer(data=request.data)
    serializer.is_valid(raise_exception=True)
    serializer.save()
    headers = self.get_success_headers(serializer.data)
    return Response(
        serializer.data.get("key"), status=status.HTTP_201_CREATED, headers=headers
    )

JiraCommentAlertView ¤

Bases: CreateAPIView[Any]

Methods:

  • post –

    Allow to send a message in Slack when a comment in a Jira ticket is created or modified.

post ¤

post(request: Request, *args: Never, **kwargs: Never) -> Response

Allow to send a message in Slack when a comment in a Jira ticket is created or modified.

Authentication: Jira signs the raw request body with settings.RAID_JIRA_WEBHOOK_SECRET and sends the HMAC-SHA256 signature in the X-Hub-Signature: sha256=<hex> header. Configure the same secret in the Atlassian webhook "Secret" field.

Source code in src/firefighter/raid/views/__init__.py
def post(self, request: Request, *args: Never, **kwargs: Never) -> Response:
    """Allow to send a message in Slack when a comment in a Jira ticket is created or modified.

    Authentication: Jira signs the raw request body with
    `settings.RAID_JIRA_WEBHOOK_SECRET` and sends the HMAC-SHA256
    signature in the `X-Hub-Signature: sha256=<hex>` header. Configure
    the same secret in the Atlassian webhook "Secret" field.
    """
    serializer = self.get_serializer(data=request.data)
    serializer.is_valid(raise_exception=True)
    serializer.save()

    return Response()

JiraUpdateAlertView ¤

Bases: CreateAPIView[Any]

Methods:

  • post –

    Allow to send a message in Slack when some fields ("Priority", "project", "description", "status") of a Jira ticket are updated.

post ¤

post(request: Request, *args: Never, **kwargs: Never) -> Response

Allow to send a message in Slack when some fields ("Priority", "project", "description", "status") of a Jira ticket are updated.

Authentication: Jira signs the raw request body with settings.RAID_JIRA_WEBHOOK_SECRET and sends the HMAC-SHA256 signature in the X-Hub-Signature: sha256=<hex> header. Configure the same secret in the Atlassian webhook "Secret" field.

Source code in src/firefighter/raid/views/__init__.py
def post(self, request: Request, *args: Never, **kwargs: Never) -> Response:
    """Allow to send a message in Slack when some fields ("Priority", "project", "description", "status") of a Jira ticket are updated.

    Authentication: Jira signs the raw request body with
    `settings.RAID_JIRA_WEBHOOK_SECRET` and sends the HMAC-SHA256
    signature in the `X-Hub-Signature: sha256=<hex>` header. Configure
    the same secret in the Atlassian webhook "Secret" field.
    """
    serializer = self.get_serializer(data=request.data)
    serializer.is_valid(raise_exception=True)
    serializer.save()

    return Response()