jira_app ¤

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

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

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

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]

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

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

def create_postmortem_for_incident(
    self,
    incident: Incident,
    created_by: User | None = None,
) -> JiraPostMortem:
    """Create a Jira post-mortem for an incident.

    Args:
        incident: Incident to create post-mortem for
        created_by: User creating the post-mortem

    Returns:
        JiraPostMortem instance

    Raises:
        ValueError: If incident already has a Jira post-mortem
        JiraAPIError: If Jira API call fails
    """
    if hasattr(incident, "jira_postmortem_for"):
        error_msg = f"Incident #{incident.id} already has a Jira post-mortem"
        raise ValueError(error_msg)

    logger.info(f"Creating Jira post-mortem for incident #{incident.id}")

    # Prefetch incident updates and jira_ticket for timeline and parent link
    from firefighter.incidents.models.incident import Incident  # noqa: PLC0415

    incident = (
        Incident.objects.select_related("priority", "environment", "jira_ticket")
        .prefetch_related("incidentupdate_set")
        .get(pk=incident.pk)
    )

    # Generate content from templates
    fields = self._generate_issue_fields(incident)

    # Get parent issue key from RAID Jira ticket if available
    parent_issue_key = None
    if hasattr(incident, "jira_ticket") and incident.jira_ticket:
        parent_issue_key = incident.jira_ticket.key

    # Create Jira issue with optional parent link
    jira_issue = self.client.create_postmortem_issue(
        project_key=self.project_key,
        issue_type=self.issue_type,
        fields=fields,
        parent_issue_key=parent_issue_key,
    )

    # Assign to incident commander if available
    commander = (
        incident.roles_set.select_related("user__jira_user", "role_type")
        .filter(role_type__slug="commander")
        .first()
    )
    if commander:
        jira_user = getattr(commander.user, "jira_user", None)
        if jira_user is None:
            try:
                jira_user = self.client.get_jira_user_from_user(commander.user)
            except (JiraUserNotFoundError, JiraUserDatabaseError) as exc:
                logger.warning(
                    "Unable to fetch Jira user for commander %s: %s",
                    commander.user_id,
                    exc,
                )
        if jira_user is not None:
            assigned = self.client.assign_issue(
                issue_key=jira_issue["key"],
                account_id=jira_user.id,
            )
            if assigned:
                logger.info(
                    "Assigned post-mortem %s to commander %s",
                    jira_issue["key"],
                    commander.user.username,
                )

    # Create JiraPostMortem record
    jira_postmortem = JiraPostMortem.objects.create(
        incident=incident,
        jira_issue_key=jira_issue["key"],
        jira_issue_id=jira_issue["id"],
        created_by=created_by,
    )

    logger.info(
        f"Created Jira post-mortem {jira_postmortem.jira_issue_key} "
        f"for incident #{incident.id}"
    )

    return jira_postmortem