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={},
        )

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:

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, PLR0917
    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,  # noqa: FBT001
    is_seller_in_golden_list: bool | None,  # noqa: FBT001
    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_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)
    """
    if "_status" not in updated_fields:
        return

    if not hasattr(incident, "jira_ticket") or incident.jira_ticket is None:
        logger.warning(
            f"Trying to close Jira ticket for incident {incident.id} but no Jira ticket found"
        )
        return

    # Determine if we should close the ticket based on status and priority
    should_close = False

    if incident_update.status == IncidentStatus.CLOSED:
        # Always close on CLOSED regardless of priority
        should_close = True
    elif incident_update.status == IncidentStatus.MITIGATED:
        # Only close on MITIGATED if incident doesn't need postmortem (P3+)
        should_close = not incident.needs_postmortem

    # POST_MORTEM status never closes the ticket - it stays open during PM phase

    if should_close:
        status_label = incident_update.status.label if incident_update.status else "Unknown"
        logger.info(f"Closing Jira ticket for incident {incident.id} (status: {status_label})")
        client.close_issue(issue_id=incident.jira_ticket.id)

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

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