jira_app ¤

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
    """
    watchers = self.jira.watchers(jira_issue_id).raw.get("watchers")
    if len(watchers) == 0:
        logger.warning(
            "Watchers not found for jira_account_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={},
        )