jira_app

admin

client

JiraClient

JiraClient()

Source code in src/firefighter/jira_app/client.py

def __init__(self) -> None:
    self.url = RAID_JIRA_API_URL

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[JiraAPIUser]

Fetch watchers for a specific Jira ticket from Jira API.

Parameters:

• jira_issue_id (str | int) –

  Jira issue id

Raises:

• ValueError –

  Empty issue id

Returns:

• list ( User ) –

  List of Jira users object

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
    """
    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

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

models

JiraIssue

Bases: Model

Jira issue model.

JiraUser

Bases: Model

Jira user model. It maps an User with the Jira account id.

tasks

sync_users_jira

utils

pythonic_keys

pythonic_keys(d: T) -> T

Converts camelCase keys in a dict or list of dict to snake_case. Works recursively.

Source code in src/firefighter/jira_app/utils.py

def pythonic_keys(d: T) -> T:
    """Converts camelCase keys in a dict or list of dict to snake_case. Works recursively."""
    if isinstance(d, dict):
        new_d = {}
        for key, value in d.items():
            new_key = _snake_case_key(key)
            new_d[new_key] = pythonic_keys(value)
        return new_d  # type: ignore[return-value]
    if isinstance(d, list):
        return [pythonic_keys(v) for v in d]  # type: ignore[return-value]
    return d