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

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)
        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_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:
        logger.info(f"Issue {issue_id} is already closed. Not closing again.")

    # Apply transitions
    # XXX Better error handling
    for transition in transitions_to_apply:
        logger.debug(f"Running transition: {transition}")
        self.jira.transition_issue(
            issue=issue_id,
            transition=transition,
            fields={},
        )

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,
    }

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

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 MITIGATING, Jira must go through two steps:
    # "Pending resolution" then "in progress".
    if incident_update.status == IncidentStatus.MITIGATING:
        all_steps_succeeded = True
        for step in (JIRA_STATUS_PENDING_RESOLUTION, JIRA_STATUS_IN_PROGRESS):
            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"),
        )

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. 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 send a message in Slack when a comment in a Jira ticket is created or modified.
    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()

    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. 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 send a message in Slack when some fields ("Priority", "project", "description", "status") of a Jira ticket are updated.
    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()

    return Response()