confluence

admin

client

ConfluenceClient

ConfluenceClient(base_url: str, username: str, api_key: str)

Bases: HttpClient

Helper methods for Confluence API.

Should not be used directly, be used by firefighter.confluence.service.ConfluenceService.

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

def __init__(self, base_url: str, username: str, api_key: str):
    super().__init__(client_kwargs={"auth": (username, api_key)})
    self.base_url_api = base_url
    """Confluence API base URL. (with `/wiki/rest/api`)"""
    self.base_url = self.base_url_api.removesuffix("/rest/api")
    """Confluence base URL. (with `/wiki`, without `/rest/api`)"""

base_url instance-attribute

base_url = self.base_url_api.removesuffix('/rest/api')

Confluence base URL. (with /wiki, without /rest/api)

base_url_api instance-attribute

base_url_api: str = base_url

Confluence API base URL. (with /wiki/rest/api)

management

commands

sort_postmortems

sort_runbooks

sync_postmortems

sync_runbooks

models

ConfluencePage

Bases: Model

Represents a Confluence page.

PostMortem

Bases: ConfluencePage

Represents a Confluence PostMortem page.

RunbookFilterSet

Bases: FilterSet

Set of filters for Runbooks.

RunbookManager

Bases: Manager['Runbook']

search staticmethod

search(queryset: QuerySet[Runbook] | None, search_term: str) -> tuple[QuerySet[Runbook], bool]

Parameters:

• queryset (QuerySet[Runbook] | None) –

  Queryset to search in. If None, search in all incidents. The Queryset allows to search on a subset of Runbooks (already filtered).

• search_term (str) –

  Search term.

Returns:

• tuple[QuerySet[Runbook], bool] –

  tuple[QuerySet[Runbook], bool]: Queryset of Runbooks matching the search term, and a boolean indicating if the search may contain duplicates objects.

Source code in src/firefighter/confluence/models.py

@staticmethod
def search(
    queryset: QuerySet[Runbook] | None, search_term: str
) -> tuple[QuerySet[Runbook], bool]:
    """Args:
        queryset (QuerySet[Runbook] | None): Queryset to search in. If None, search in all incidents. The Queryset allows to search on a subset of Runbooks (already filtered).
        search_term (str): Search term.

    Returns:
        tuple[QuerySet[Runbook], bool]: Queryset of Runbooks matching the search term, and a boolean indicating if the search may contain duplicates objects.
    """
    if queryset is None:
        queryset = Runbook.objects.all()

    # If not search, return the original queryset
    if search_term is None or search_term.strip() == "":
        return queryset, False

    # XXX Improve search performance and relevance
    vector = (
        SearchVector("name", config="english", weight="A")
        + SearchVector("service_name", config="english", weight="B")
        + SearchVector("body_storage", config="english", weight="C")
    )
    query = SearchQuery(search_term, config="english", search_type="websearch")
    queryset = (
        queryset.annotate(rank=SearchRank(vector, query))
        .filter(rank__gte=0.05)
        .order_by("-rank")
    )

    return queryset, False

service

ConfluenceService

ConfluenceService()

Provide methods to interact with Confluence, without regards to the underlying library or HTTP client used.

TODO: The service should provide Pythonic errors and return types, and should not leak the underlying library or Confluence errors. It is not a priority for now, as the Confluence API is not used in critical paths.

Source code in src/firefighter/confluence/service.py

def __init__(self) -> None:
    self._client = None

base_url class-attribute instance-attribute

base_url: str = settings.CONFLUENCE_URL.removesuffix('/rest/api')

Confluence base URL. (with /wiki, without /rest/api)

create_page

create_page(title: str, parent_id: ConfluencePageId, body: str, space: str | None = None) -> ConfluencePage

TODO: Errors.

Parameters:

• title (str) –

  Title of the page to create.

• parent_id (ConfluencePageId) –

  Where to create the page. Will be a child of this page.

• body (str) –

  Body of the page.

• space (str | None, default: None ) –

  Space to create the page in. If None, defaults to theService's default space. Defaults to None.

Returns:

• ConfluencePage ( ConfluencePage ) –

  The newly created page.

Source code in src/firefighter/confluence/service.py

def create_page(
    self,
    title: str,
    parent_id: ConfluencePageId,
    body: str,
    space: str | None = None,
) -> ConfluencePage:
    """TODO: Errors.

    Args:
        title (str): Title of the page to create.
        parent_id (ConfluencePageId): Where to create the page. Will be a child of this page.
        body (str): Body of the page.
        space (str | None, optional): Space to create the page in. If None, defaults to theService's default space. Defaults to None.

    Returns:
        ConfluencePage: The newly created page.
    """
    if not space:
        space = self.DEFAULT_SPACE
    return self.client.create_page(title, parent_id, space, body).json()

create_postmortem

create_postmortem(title: str) -> None | ConfluencePage

Create a PostMortem page.

Parameters:

• title (str) –

  Title of the PostMortem page.

Returns:

• None | ConfluencePage –

  None | ConfluencePage: The newly created page, or None if it failed.

Source code in src/firefighter/confluence/service.py

def create_postmortem(self, title: str) -> None | ConfluencePage:
    """Create a PostMortem page.

    Args:
        title (str): Title of the PostMortem page.

    Returns:
        None | ConfluencePage: The newly created page, or None if it failed.
    """
    # Get the body from the template page
    postmortem_template = (
        self.client.get_page(self.POSTMORTEM_TEMPLATE_PAGE_ID)
    ).json()
    body = get_in(postmortem_template, "body.storage.value")
    logger.debug(body)

    # Create the postmortem with the template body
    page_created: ConfluencePage = self.create_page(
        title, self.POSTMORTEM_FOLDER_ID, body
    )

    logger.debug("Postmortem page id: %s ", page_created.get("id"))
    if "statusCode" in page_created:
        logger.error(
            f"Can't create postmortem page with title {title}: {page_created}"
        )
        return None
    logger.debug(page_created)
    return page_created

get_page

get_page(page_id: ConfluencePageId) -> ConfluencePage

TODO: Errors.

Parameters:

• page_id (ConfluencePageId) –

  ID of the page to get.

Returns:

• ConfluencePage ( ConfluencePage ) –

  The page.

Source code in src/firefighter/confluence/service.py

def get_page(self, page_id: ConfluencePageId) -> ConfluencePage:
    """TODO: Errors.

    Args:
        page_id (ConfluencePageId): ID of the page to get.

    Returns:
        ConfluencePage: The page.
    """
    return self.client.get_page(page_id).json()

get_page_children_pages

get_page_children_pages(page_id: ConfluencePageId, *args: Any, **kwargs: Any) -> list[ConfluencePage]

Get all children pages of a given page. TODO: testing.

Parameters:

• page_id (ConfluencePageId) –

  ID of the page to get children pages from.

• *args (Any, default: () ) –

  Arguments to pass to Confluence.

• **kwargs (Any, default: {} ) –

  Keyword arguments to pass to Confluence.

Returns:

• list[ConfluencePage] –

  list[ConfluencePage]: list of children pages.

Source code in src/firefighter/confluence/service.py

def get_page_children_pages(
    self, page_id: ConfluencePageId, *args: Any, **kwargs: Any
) -> list[ConfluencePage]:
    """Get all children pages of a given page.
    TODO: testing.

    Args:
        page_id (ConfluencePageId): ID of the page to get children pages from.
        *args: Arguments to pass to Confluence.
        **kwargs: Keyword arguments to pass to Confluence.

    Returns:
        list[ConfluencePage]: list of children pages.
    """
    return list(self.client.get_page_children_pages(page_id, *args, **kwargs))

get_page_descendant_pages

get_page_descendant_pages(page_id: ConfluencePageId, *args: Any, **kwargs: Any) -> list[ConfluencePage]

Get all descendant (nested children) pages of a given page. TODO: Pagination support and testing.

Parameters:

• page_id (ConfluencePageId) –

  ID of the page to get children pages from.

• *args (Any, default: () ) –

  Arguments to pass to Confluence.

• **kwargs (Any, default: {} ) –

  Keyword arguments to pass to Confluence.

Returns:

• list[ConfluencePage] –

  list[ConfluencePage]: list of children pages.

Source code in src/firefighter/confluence/service.py

def get_page_descendant_pages(
    self, page_id: ConfluencePageId, *args: Any, **kwargs: Any
) -> list[ConfluencePage]:
    """Get all descendant (nested children) pages of a given page.
    TODO: Pagination support and testing.

    Args:
        page_id (ConfluencePageId): ID of the page to get children pages from.
        *args: Arguments to pass to Confluence.
        **kwargs: Keyword arguments to pass to Confluence.

    Returns:
        list[ConfluencePage]: list of children pages.
    """
    return self.client.get_page_descendant_pages(page_id, *args, **kwargs).json()[
        "results"
    ]

move_page

move_page(
    page_id: ConfluencePageId,
    target_page_id: ConfluencePageId,
    position: Literal["before", "after", "append"] = "append",
    *,
    dry_run: bool = False
) -> None | ConfluencePage

Parameters:

• page_id (ConfluencePageId) –

  ID of the page to move.

• target_page_id (ConfluencePageId) –

  ID of the parent page to move the page in relation to.

• position (Literal['before', 'after', 'append'], default: 'append' ) –

  Where should the page be moved, in relation to the target page. Defaults to "append".

• dry_run (bool, default: False ) –

  If True, will log instead of moving. Defaults to False.

Returns:

• None | ConfluencePage –

  None | ConfluencePage: The moved page, or None if it failed.

Source code in src/firefighter/confluence/service.py

def move_page(
    self,
    page_id: ConfluencePageId,
    target_page_id: ConfluencePageId,
    position: Literal["before", "after", "append"] = "append",
    *,
    dry_run: bool = False,
) -> None | ConfluencePage:
    """Args:
        page_id (ConfluencePageId): ID of the page to move.
        target_page_id (ConfluencePageId): ID of the parent page to move the page in relation to.
        position (Literal["before", "after", "append"], optional): Where should the page be moved, in relation to the target page. Defaults to "append".
        dry_run (bool, optional): If True, will log instead of moving. Defaults to False.

    Returns:
        None | ConfluencePage: The moved page, or None if it failed.
    """
    if dry_run:
        logger.info(
            f"Would have moved {page_id} {'into' if position == 'append' else position} {target_page_id} if not dry run"
        )
        return None

    res = self.client.move_page(page_id, target_page_id, position)
    logger.info(
        f"Moved {page_id} {'into' if position == 'append' else position} {target_page_id}"
    )
    if res.status_code == 200:
        return res.json()
    logger.error(f"Error moving page {page_id} {res.status_code} {res.text}")
    return None

parse_confluence_page

parse_confluence_page(pm: dict[str, Any] | ConfluencePage) -> PageInfo

Helper to parse a Confluence page into a PageInfo object.

Parameters:

• pm (dict[str, Any] | ConfluencePage) –

  ConfluencePage from the API, or a dict with the same structure.

Returns:

• PageInfo ( PageInfo ) –

  The parsed page details that are present on firefighter.confluence.models.ConfluencePage.

Raises:

• TypeError –

  If one of page_id, page_title, page_url or page_edit_url are not strings.

Source code in src/firefighter/confluence/service.py

def parse_confluence_page(self, pm: dict[str, Any] | ConfluencePage) -> PageInfo:
    """Helper to parse a Confluence page into a PageInfo object.

    Args:
        pm (dict[str, Any] | ConfluencePage): ConfluencePage from the API, or a dict with the same structure.

    Returns:
        PageInfo: The parsed page details that are present on [firefighter.confluence.models.ConfluencePage][].

    Raises:
        TypeError: If one of page_id, page_title, page_url or page_edit_url are not strings.
    """
    page_url = f'{self.base_url}{get_in(pm, "_links.webui")}'
    page_edit_url = f'{self.base_url}{get_in(pm, "_links.editui")}'

    page_id = str(pm.get("id"))
    page_title = pm.get("title")
    if not isinstance(page_id, str):
        raise TypeError("page_id is not a string")
    if not isinstance(page_title, str):
        raise TypeError("page_title is not a string")
    if not isinstance(page_url, str):
        raise TypeError("page_url is not a string")
    if not isinstance(page_edit_url, str):
        raise TypeError("page_edit_url is not a string")
    return {
        "name": page_title,
        "page_id": page_id,
        "page_url": page_url,
        "page_edit_url": page_edit_url,
    }

sort_pages

sort_pages(page_ids: list[tuple[ConfluencePageId, ConfluencePage]], *, dry_run: bool = False) -> None

Sort pages according to the list order.

Be careful: - this will move pages around, and will not check if the pages are already in the right order. - this will not check that the pages are in the same parent. - this will perform n moves, where n is the number of pages in the list.

Parameters:

• page_ids (list[tuple[ConfluencePageId, ConfluencePage]]) –

  Sorted list of (PageId, Page) tuples.

• dry_run (bool, default: False ) –

  If True, will log instead of sorting. Defaults to False.

Source code in src/firefighter/confluence/service.py

def sort_pages(
    self,
    page_ids: list[tuple[ConfluencePageId, ConfluencePage]],
    *,
    dry_run: bool = False,
) -> None:
    """Sort pages according to the list order.

    **Be careful**:
    - this will move pages around, and will not check if the pages are already in the right order.
    - this will not check that the pages are in the same parent.
    - this will perform n moves, where n is the number of pages in the list.

    Args:
        page_ids (list[tuple[ConfluencePageId, ConfluencePage]]): Sorted list of (PageId, Page) tuples.
        dry_run (bool, optional): If True, will log instead of sorting. Defaults to False.
    """
    for index, page in enumerate(page_ids):
        if index == 0:
            continue

        target_id = page_ids[index - 1][0]
        page_id = page[0]
        self.move_page(
            page_id,
            target_id,
            position="after",
            dry_run=dry_run,
        )

update_oncall_page

update_oncall_page(users: dict[str, User]) -> bool

Update the Confluence list of On-Call users, if the page needs to be updated. Users should have a SlackUser AND a PagerDutyUser associated.

Parameters:

• users (list[User]) –

  list of Users to update the page with.

Returns:

• bool ( bool ) –

  has the page been updated?

Source code in src/firefighter/confluence/service.py

def update_oncall_page(self, users: dict[str, User]) -> bool:
    """Update the Confluence list of On-Call users, if the page needs to be updated.
    Users should have a SlackUser AND a PagerDutyUser associated.

    Args:
        users (list[User]): list of Users to update the page with.

    Returns:
        bool: has the page been updated?
    """
    if not settings.CONFLUENCE_ON_CALL_PAGE_ID:
        logger.info("No Confluence OnCall page ID, skipping.")
        return False

    content = (self.client.get_page(settings.CONFLUENCE_ON_CALL_PAGE_ID)).json()

    page_version = get_in(content, "version.number")
    if not page_version:
        logger.error("No page version in Confluence page! %s", content)
        return False
    page_version += 1
    page_body = self._build_page_from_django_template(
        "oncall_team.xml",
        context={
            "users": users.items(),
            "oncall_page_link": urljoin(
                settings.BASE_URL, reverse("pagerduty:oncall-list")
            ),
        },
    )
    logger.debug("Confluence OnCall page body: %s", page_body)
    if get_in(content, "body.storage.value") != page_body:
        res = self.client.update_page(
            content.get("id"),
            content.get("type"),
            content.get("title"),
            page_body,
            page_version,
        )
        if res.status_code != 200:
            logger.error("Can't update OnCall page: %s", res.json())
            return False
        return True

    logger.info("Confluence OnCall page is up to date, and was not updated.")
    return False

signals

incident_updated

tasks

archive_postmortems

archive_and_sort_postmortems

archive_and_sort_postmortems(*, dry_run: bool = True) -> None

Archive postmortems that are older than 2 weeks, for incidents that are closed.

Parameters:

• dry_run (bool, default: True ) –

  Should actions be performed? If True, Confluence will be accessed in read-only mode. Defaults to False.

Source code in src/firefighter/confluence/tasks/archive_postmortems.py

@shared_task(
    name="confluence.archive_and_sort_postmortems",
    soft_time_limit=60 * 5,
    time_limit=60 * 7,
)
def archive_and_sort_postmortems(*, dry_run: bool = True) -> None:
    """Archive postmortems that are older than 2 weeks, for incidents that are closed.

    Args:
        dry_run (bool, optional): Should actions be performed? If True, Confluence will be accessed in read-only mode. Defaults to False.
    """
    logger.info(f"Running in {'dry run' if dry_run else 'real'} mode")

    # 1. Get top level pages
    pm_to_sort, quarter_bins = _get_top_level_pages(
        confluence_service.POSTMORTEM_FOLDER_ID
    )

    # 2. Create current quarter folder if it does not exist
    create_current_bin_if_needed(quarter_bins, dry_run=dry_run)

    # 3. Archive postmortems in root into their quarter folder (if appropriate)
    _archive_postmortems(pm_to_sort, quarter_bins, dry_run=dry_run)

    # 4. Move archived postmortems in the right quarter folder if there is any error
    # Not needed atm

    # 5. Sort postmortems in each quarter folder
    sort_postmortems_in_bins(quarter_bins, dry_run=dry_run)

sort_postmortems_in_bins

sort_postmortems_in_bins(quarter_bins: dict[str, int], *, dry_run: bool = False) -> None

Sort postmortems in each quarter archive folder.

Parameters:

• quarter_bins (dict[str, int]) –

  description

• dry_run (bool, default: False ) –

  description. Defaults to False.

Source code in src/firefighter/confluence/tasks/archive_postmortems.py

def sort_postmortems_in_bins(
    quarter_bins: dict[str, int], *, dry_run: bool = False
) -> None:
    """Sort postmortems in each quarter archive folder.

    Args:
        quarter_bins (dict[str, int]): _description_
        dry_run (bool, optional): _description_. Defaults to False.
    """
    for quarter, quarter_page_id in quarter_bins.items():
        # Get children per archive page
        children: list[ConfluencePage] = confluence_service.get_page_children_pages(
            quarter_page_id, expand=""
        )
        logger.info(f"Found {len(children)} postmortems in {quarter}")
        quarter_children: list[tuple[int, datetime, ConfluencePage]] = []

        for page in children:
            title: str = page["title"]
            page_id: int = int(page["id"])

            date_, fmt = parse_postmortem_title(title)
            if date_ and fmt:
                quarter_children.append((page_id, date_, page))

        sorted_children = sorted(quarter_children, key=lambda x: x[1], reverse=True)
        # Check if already sorted
        if sorted_children == quarter_children:
            logger.debug(f"Quarter {quarter} is already sorted. Skipping")
            continue

        logger.info(f"Quarter {quarter} is not sorted. Sorting")

        clean_sorted: list[tuple[ConfluencePageId, ConfluencePage]] = [
            (x[0], x[2]) for x in sorted_children
        ]

        confluence_service.sort_pages(clean_sorted, dry_run=dry_run)

        logger.info(f"Fetched {len(quarter_children)} postmortems")

sort_runbooks

sort_runbooks

sort_runbooks(*, dry_run: bool = False) -> None

Sort runbooks in Confluence.

Parameters:

• dry_run (bool, default: False ) –

  Should actions be performed? If True, Confluence will be accessed in read-only mode. Defaults to False.

Source code in src/firefighter/confluence/tasks/sort_runbooks.py

@shared_task(name="confluence.sort_runbooks", soft_time_limit=60 * 5, time_limit=60 * 7)
def sort_runbooks(*, dry_run: bool = False) -> None:
    """Sort runbooks in Confluence.

    Args:
        dry_run (bool, optional): Should actions be performed? If True, Confluence will be accessed in read-only mode. Defaults to False.
    """
    logger.info(f"Running in {'dry run' if dry_run else 'real'} mode")

    # 1. Get top level pages
    folders = _get_top_level_pages(confluence_service.RUNBOOKS_FOLDER_ID)

    # 2. Sort runbooks in each folder
    sort_runbooks_in_folders(folders, dry_run=dry_run)

sort_runbooks_in_folders

sort_runbooks_in_folders(folders: list[ConfluencePageId], *, dry_run: bool = False) -> None

Sort runbooks in each folder.

Parameters:

• folders (list[ConfluencePageId]) –

  List of folders to sort runbooks in.

• dry_run (bool, default: False ) –

  Do not perform the sort. Defaults to False.

Source code in src/firefighter/confluence/tasks/sort_runbooks.py

def sort_runbooks_in_folders(
    folders: list[ConfluencePageId], *, dry_run: bool = False
) -> None:
    """Sort runbooks in each folder.

    Args:
        folders (list[ConfluencePageId]): List of folders to sort runbooks in.
        dry_run (bool, optional): Do not perform the sort. Defaults to False.
    """
    for folder_page_id in folders:
        # Get children per archive page
        children: list[ConfluencePage] = confluence_service.get_page_children_pages(
            folder_page_id, expand=""
        )

        folder_children: list[tuple[ConfluencePageId, str, ConfluencePage]] = []

        for page in children:
            title: str = page["title"]
            page_id: ConfluencePageId = page["id"]

            runbook_name_clean = parse_runbook_title(title)
            if runbook_name_clean:
                folder_children.append((page_id, runbook_name_clean, page))

        sorted_children = sorted(folder_children, key=lambda x: x[1].lower())
        # Check if already sorted
        if sorted_children == folder_children:
            logger.debug(f"Folder {folder_page_id} is already sorted. Skipping")
            continue
        clean_sorted_children: list[tuple[ConfluencePageId, ConfluencePage]] = [
            (x[0], x[2]) for x in sorted_children
        ]
        logger.info(f"Folder {folder_page_id} is not sorted. Sorting")
        confluence_service.sort_pages(clean_sorted_children, dry_run=dry_run)

sync_pages_content

sync_postmortems

sync_runbooks

urls

utils

ConfluencePageId module-attribute

ConfluencePageId: TypeAlias = int | str

Alias of int | str

PageInfo

Bases: TypedDict

Dict mapped to mandatory fields for a firefighter.confluence.models.ConfluencePage.

parse_postmortem_title

parse_postmortem_title(title: str) -> tuple[datetime, str] | tuple[None, None]

Parse postmortem title and return the date and quarter associated to this date, in the format YYYYQX.

Parameters:

• title (str) –

  Title of the postmortem, in the format #YYYYMMDD-- (SEV/P)