mkdocs_to_confluence.plugin

MkDocs plugin for publishing to Confluence.

BearerAuth

BearerAuth(token)

Bases: AuthBase

Attaches OAuth Bearer Token Authentication to the given Request object.

Store the OAuth bearer token used for authentication.

Parameters:

Name	Type	Description	Default
token	str	OAuth access token issued by Atlassian.	required

Source code in src/mkdocs_to_confluence/plugin.py

def __init__(self, token):
    """Store the OAuth bearer token used for authentication.

    Args:
        token (str): OAuth access token issued by Atlassian.

    """
    self.token = token

DummyFile

File-like object that discards all writes.

write

write(x)

Discard written content.

Source code in src/mkdocs_to_confluence/plugin.py

def write(self, x):
    """Discard written content."""
    pass

MkdocsWithConfluence

MkdocsWithConfluence()

Bases: BasePlugin

MkDocs plugin to publish documentation to Confluence.

Initialize plugin with default settings.

Source code in src/mkdocs_to_confluence/plugin.py

def __init__(self):
    """Initialize plugin with default settings."""
    self.enabled = True
    self.confluence_renderer = ConfluenceRenderer()
    self.confluence_mistune = mistune.Markdown(renderer=self.confluence_renderer)
    self.flen = 1
    self.session = requests.Session()
    self.page_attachments = {}
    self.dryrun = False
    self.export_only = False
    self.exporter = None
    self.synced_pages: set[str] = set()
    self.page_status: dict[str, str] = {}  # Track page statuses for summary

on_nav

on_nav(nav, config, files)

Build navigation structure from MkDocs nav.

Parameters:

Name	Type	Description	Default
nav	Navigation	Navigation object produced by MkDocs.	required
config	Config	Active MkDocs configuration.	required
files	Files	Collection of files included in the build.	required

Source code in src/mkdocs_to_confluence/plugin.py

def on_nav(self, nav, config, files):
    """Build navigation structure from MkDocs nav.

    Args:
        nav (mkdocs.structure.nav.Navigation): Navigation object produced by MkDocs.
        config (mkdocs.config.base.Config): Active MkDocs configuration.
        files (mkdocs.structure.files.Files): Collection of files included in the build.

    """
    MkdocsWithConfluence.tab_nav = []
    navigation_items = nav.__repr__()

    for n in navigation_items.split("\n"):
        leading_spaces = len(n) - len(n.lstrip(" "))
        spaces = leading_spaces * " "
        if "Page" in n:
            try:
                self.page_title = self.__get_page_title(n)
                if self.page_title is None:
                    raise AttributeError
            except AttributeError:
                self.page_local_path = self.__get_page_url(n)
                logger.warning(
                    f"Page from path {self.page_local_path} has no "
                    f"entity in the mkdocs.yml nav section. It will be uploaded "
                    f"to Confluence, but you may not see it on the web server!"
                )
                self.page_local_name = self.__get_page_name(n)
                self.page_title = self.page_local_name

            p = spaces + self.page_title
            MkdocsWithConfluence.tab_nav.append(p)
        if "Section" in n:
            try:
                self.section_title = self.__get_section_title(n)
                if self.section_title is None:
                    raise AttributeError
            except AttributeError:
                self.section_local_path = self.__get_page_url(n)
                logger.warning(
                    f"Section from path {self.section_local_path} has no "
                    f"entity in the mkdocs.yml nav section. It will be uploaded "
                    f"to Confluence, but you may not see it on the web server!"
                )
                self.section_local_name = self.__get_section_title(n)
                self.section_title = self.section_local_name
            s = spaces + self.section_title
            MkdocsWithConfluence.tab_nav.append(s)

on_files

on_files(files, config)

Count documentation pages.

Parameters:

Name	Type	Description	Default
files	Files	Files selected for the build.	required
config	Config	Active MkDocs configuration.	required

Source code in src/mkdocs_to_confluence/plugin.py

def on_files(self, files, config):
    """Count documentation pages.

    Args:
        files (mkdocs.structure.files.Files): Files selected for the build.
        config (mkdocs.config.base.Config): Active MkDocs configuration.

    """
    pages = files.documentation_pages()
    self.flen = len(pages)
    logger.info(f"Number of files in directory tree: {self.flen}")
    if self.flen == 0:
        logger.error("No documentation pages in directory tree, please add at least one!")

on_config

on_config(config)

Configure plugin based on environment and settings.

Parameters:

Name	Type	Description	Default
config	Config	Active MkDocs configuration being initialized.	required

Source code in src/mkdocs_to_confluence/plugin.py

def on_config(self, config):
    """Configure plugin based on environment and settings.

    Args:
        config (mkdocs.config.base.Config): Active MkDocs configuration being initialized.

    """
    # Print version
    try:
        plugin_version = version("mkdocs-to-confluence")
        logger.info(f"Mkdocs With Confluence v{plugin_version}")
    except Exception:
        logger.info("Mkdocs With Confluence (version unknown)")

    # Allow environment variables to override debug and verbose settings
    if os.environ.get("CONFLUENCE_DEBUG", "").lower() in ("true", "1"):
        self.config["debug"] = True
    elif os.environ.get("CONFLUENCE_DEBUG", "").lower() in ("false", "0"):
        self.config["debug"] = False

    if os.environ.get("CONFLUENCE_VERBOSE", "").lower() in ("true", "1"):
        self.config["verbose"] = True
    elif os.environ.get("CONFLUENCE_VERBOSE", "").lower() in ("false", "0"):
        self.config["verbose"] = False

    # Always show configuration status for troubleshooting (without exposing credentials)
    logger.info("Configuration check:")
    logger.info(f"  - username configured: {bool(self.config.get('username'))}")
    logger.info(f"  - api_token configured: {bool(self.config.get('api_token'))}")
    logger.info(f"  - password configured: {bool(self.config.get('password'))}")
    logger.info(f"  - auth_type: {self.config.get('auth_type', 'basic')}")

    # Handle export_only and dryrun modes
    if self.config["export_only"] and self.config["dryrun"]:
        logger.warning(
            "Both export_only and dryrun are enabled. "
            "Using export_only mode (no Confluence connection)."
        )
        self.export_only = True
        self.dryrun = False
        export_dir = Path(self.config["export_dir"])
        self.exporter = ConfluenceExporter(export_dir)
        logger.info(f"Mkdocs With Confluence: EXPORT ONLY mode - Exporting to {export_dir}")
    elif self.config["export_only"]:
        logger.warning("Mkdocs With Confluence - EXPORT ONLY MODE turned ON (no Confluence connection)")
        self.export_only = True
        self.dryrun = False
        export_dir = Path(self.config["export_dir"])
        self.exporter = ConfluenceExporter(export_dir)
        logger.info(f"Mkdocs With Confluence: Exporting to {export_dir}")
    elif self.config["dryrun"]:
        logger.warning("Mkdocs With Confluence - DRYRUN MODE turned ON (read-only, no modifications)")
        self.dryrun = True
        self.export_only = False
        self.exporter = None
    else:
        self.dryrun = False
        self.export_only = False
        self.exporter = None

    # Handle conditional enabling based on environment variable
    env_name = self.config.get("enabled_if_env")
    if env_name is not None:
        # Conditional execution based on environment variable
        self.enabled = os.environ.get(env_name) == "1"
        if not self.enabled:
            logger.warning(
                "Mkdocs With Confluence: Exporting MKDOCS pages to Confluence turned OFF: "
                f"(set environment variable {env_name} to 1 to enable)"
            )
            return
        logger.info(
            "Mkdocs With Confluence: Exporting MKDOCS pages to Confluence "
            f"turned ON by var {env_name}==1!"
        )
    else:
        # No conditional - enabled by default
        logger.info("Mkdocs With Confluence: Exporting MKDOCS pages to Confluence turned ON by default!")
        self.enabled = True

on_page_markdown

on_page_markdown(
    markdown: str, page: Any, config: Any, files: Any
) -> str

Process markdown content and publish to Confluence.

Parameters:

Name	Type	Description	Default
markdown	str	Markdown generated by MkDocs for the current page.	required
page	Page	Page metadata within the nav tree.	required
config	Config	Active MkDocs configuration.	required
files	Files	Collection of build file objects.	required

Returns:

Name	Type	Description
str	str	Markdown that MkDocs should continue to render or return if syncing fails.

Source code in src/mkdocs_to_confluence/plugin.py

def on_page_markdown(self, markdown: str, page: Any, config: Any, files: Any) -> str:
    """Process markdown content and publish to Confluence.

    Args:
        markdown (str): Markdown generated by MkDocs for the current page.
        page (mkdocs.structure.pages.Page): Page metadata within the nav tree.
        config (mkdocs.config.base.Config): Active MkDocs configuration.
        files (mkdocs.structure.files.Files): Collection of build file objects.

    Returns:
        str: Markdown that MkDocs should continue to render or return if syncing fails.

    """
    MkdocsWithConfluence._id += 1

    # Set up authentication based on auth_type
    if self.config["api_token"]:
        token = self.config["api_token"]
        if self.config["auth_type"] == "bearer":
            self.session.auth = BearerAuth(token)
            if self.config["debug"]:
                logger.debug(f"Using OAuth Bearer token authentication for {self.config['username']}")
        else:
            # Use HTTP Basic Auth (default)
            # Convert None to empty string to avoid deprecation warning in requests 3.0.0
            username = self.config["username"] or ""
            if username == "":
                logger.warning("Username is not configured. Check plugin configuration or environment variables.")
            if token == "":  # nosec B105
                logger.warning("API token is not configured. Check plugin configuration or environment variables.")
            self.session.auth = (username, token)
    else:
        # Convert None to empty string to avoid deprecation warning in requests 3.0.0
        username = self.config["username"] or ""
        password = self.config["password"] or ""
        if username == "":
            logger.warning("Username is not configured. Check plugin configuration or environment variables.")
        if password == "":  # nosec B105
            logger.warning("Password is not configured. Check plugin configuration or environment variables.")
        self.session.auth = (username, password)

    if self.enabled:
        # Show progress bar only in non-verbose, non-debug mode
        if not self.config["verbose"] and not self.config["debug"]:
            progress = "#" * MkdocsWithConfluence._id
            remaining = "-" * (self.flen - MkdocsWithConfluence._id)
            logger.info(
                f"Mkdocs With Confluence: Page export progress: [{progress}{remaining}] "
                f"({MkdocsWithConfluence._id} / {self.flen})"
            )

        if self.config["debug"]:
            logger.debug(f"\nHandling Page '{page.title}' (And Parent Nav Pages if necessary):\n")
        if not all(self.config_scheme):
            logger.error("YOU HAVE EMPTY VALUES IN YOUR CONFIG. ABORTING")
            return markdown

        try:
            # Resolve parent page hierarchy
            parent_chain = self._resolve_page_parents(page)

            # Extract attachments from markdown
            attachments = self._extract_attachments(markdown)

            # Convert markdown to Confluence format
            confluence_body = self._convert_to_confluence_format(markdown, page.title)

            if self.config["debug"]:
                logger.debug(
                    f"\nUPDATING PAGE TO CONFLUENCE, DETAILS:\n"
                    f"HOST: {self.config['host_url']}\n"
                    f"SPACE: {self.config['space']}\n"
                    f"TITLE: {page.title}\n"
                    f"PARENT CHAIN: {' > '.join(parent_chain)}\n"
                    f"BODY: {confluence_body}\n"
                )

            # Sync page to Confluence or add to exporter
            if self.export_only and self.exporter:
                # Export-only mode: Add page to exporter queue (no Confluence connection)
                direct_parent = parent_chain[-1] if parent_chain else None
                root_parent = parent_chain[0] if parent_chain else None
                self.exporter.add_page(
                    title=page.title,
                    parent=direct_parent if direct_parent != root_parent else None,
                    space=self.config["space"],
                    confluence_body=confluence_body,
                    attachments=attachments,
                )
                logger.info(f"Mkdocs With Confluence: {page.title} - *QUEUED FOR EXPORT*")
            else:
                # Normal or dryrun mode: sync to Confluence (dryrun = read-only validation)
                sync_success = self._sync_page(page.title, parent_chain, confluence_body)
                if not sync_success:
                    return markdown

                if attachments:
                    self.page_attachments[page.title] = attachments

        except IndexError as e:
            if self.config["debug"]:
                logger.debug(f"ERR({e}): Exception error!")
            return markdown

    return markdown

on_post_page

on_post_page(output: str, page: Any, config: Any) -> str

Upload attachments after page is rendered.

Parameters:

Name	Type	Description	Default
output	str	Rendered HTML for the page.	required
page	Page	Page metadata used for lookup.	required
config	Config	Active MkDocs configuration.	required

Returns:

Name	Type	Description
str	str	Rendered HTML that MkDocs should write to disk.

Source code in src/mkdocs_to_confluence/plugin.py

def on_post_page(self, output: str, page: Any, config: Any) -> str:
    """Upload attachments after page is rendered.

    Args:
        output (str): Rendered HTML for the page.
        page (mkdocs.structure.pages.Page): Page metadata used for lookup.
        config (mkdocs.config.base.Config): Active MkDocs configuration.

    Returns:
        str: Rendered HTML that MkDocs should write to disk.

    """
    # Skip attachment uploads in export_only mode (no Confluence connection)
    if self.export_only:
        return output

    site_dir = config.get("site_dir")
    attachments = self.page_attachments.get(page.title, [])

    if self.config["debug"]:
        logger.debug(f"\nUPLOADING ATTACHMENTS TO CONFLUENCE FOR {page.title}, DETAILS:")
        logger.info(f"FILES: {attachments}  \n")

    for attachment in attachments:
        if self.config["debug"]:
            logger.debug(f"looking for {attachment} in {site_dir}")
        for p in Path(site_dir).rglob(f"*{attachment}"):
            if self.dryrun:
                logger.info(f"  * Attachment: {p.name} - *WOULD UPLOAD* (dryrun)")
            else:
                self.add_or_update_attachment(page.title, str(p))
    return output

on_page_content

on_page_content(
    html: str, page: Any, config: Any, files: Any
) -> str

Process HTML content.

Parameters:

Name	Type	Description	Default
html	str	Rendered HTML output.	required
page	Page	Page metadata.	required
config	Config	Active MkDocs configuration.	required
files	Files	Collection of build file objects.	required

Returns:

Name	Type	Description
str	str	HTML passed through without modification.

Source code in src/mkdocs_to_confluence/plugin.py

def on_page_content(self, html: str, page: Any, config: Any, files: Any) -> str:
    """Process HTML content.

    Args:
        html (str): Rendered HTML output.
        page (mkdocs.structure.pages.Page): Page metadata.
        config (mkdocs.config.base.Config): Active MkDocs configuration.
        files (mkdocs.structure.files.Files): Collection of build file objects.

    Returns:
        str: HTML passed through without modification.

    """
    return html

on_post_build

on_post_build(config: Any) -> None

Export all queued pages and handle orphaned pages after build completes.

Parameters:

Name	Type	Description	Default
config	Config	Active MkDocs configuration.	required

Source code in src/mkdocs_to_confluence/plugin.py

def on_post_build(self, config: Any) -> None:
    """Export all queued pages and handle orphaned pages after build completes.

    Args:
        config (mkdocs.config.base.Config): Active MkDocs configuration.

    """
    # Handle export-only mode (filesystem export, no Confluence connection)
    if self.export_only:
        if self.exporter:
            logger.info("Mkdocs With Confluence: Exporting all pages to filesystem...")
            self.exporter.export_all()
            export_dir = Path(self.config["export_dir"])
            logger.info(f"Mkdocs With Confluence: Export complete! Files saved to {export_dir.absolute()}")
        return

    # Skip orphaned page detection if plugin is disabled
    if not self.enabled:
        return

    # Get root parent page
    root_parent_name = self.config.get("parent_page_name") or self.config.get("space")
    if not root_parent_name:
        if self.config["debug"]:
            logger.debug("No root parent configured, skipping orphaned page detection")
        return

    root_parent_id = self.find_page_id(root_parent_name)
    if not root_parent_id:
        logger.warning(f"Could not find root parent page '{root_parent_name}', skipping orphaned page detection")
        return

    if self.config["debug"]:
        logger.debug(f"Checking for orphaned pages under '{root_parent_name}' (ID: {root_parent_id})")
        logger.debug(f"Synced pages during build: {sorted(self.synced_pages)}")

    # Get all pages in Confluence under the root parent
    try:
        confluence_pages = self.get_all_child_pages(root_parent_id)
    except Exception as e:
        logger.error(f"Failed to fetch Confluence pages: {e}")
        return

    if self.config["debug"]:
        confluence_titles_list = [page["title"] for page in confluence_pages]
        logger.debug(f"Pages in Confluence: {sorted(confluence_titles_list)}")

    # Find orphaned pages (in Confluence but not in synced_pages)
    confluence_titles = {page["title"] for page in confluence_pages}
    orphaned_titles = confluence_titles - self.synced_pages

    # Filter out pages in keep_pages list
    keep_pages = set(self.config.get("keep_pages", []))
    orphaned_titles = orphaned_titles - keep_pages

    if not orphaned_titles:
        logger.info("Mkdocs With Confluence: No orphaned pages found")
        return

    # Always warn about orphaned pages
    logger.warning(f"Mkdocs With Confluence: Found {len(orphaned_titles)} orphaned page(s) in Confluence:")
    for title in sorted(orphaned_titles):
        logger.warning(f"  - {title}")

    # Delete orphaned pages if cleanup is enabled (skip in dryrun)
    if self.config.get("cleanup_orphaned_pages", False):
        if self.dryrun:
            logger.info("Mkdocs With Confluence: Cleanup enabled, but DRYRUN - would delete orphaned pages")
            logger.info(f"Would delete {len(orphaned_titles)} orphaned page(s) (dryrun)")
        else:
            logger.info("Mkdocs With Confluence: Cleanup enabled, deleting orphaned pages...")
            deleted_count = 0
            for page in confluence_pages:
                if page["title"] in orphaned_titles:
                    try:
                        self.delete_page(page["id"])
                        deleted_count += 1
                    except Exception as e:
                        logger.error(f"Failed to delete page '{page['title']}': {e}")

            logger.info(f"Mkdocs With Confluence: Deleted {deleted_count} orphaned page(s)")
    else:
        if self.dryrun:
            logger.info(
                "Mkdocs With Confluence: Run with 'cleanup_orphaned_pages: true' "
                "to delete them (dryrun mode)"
            )
        else:
            logger.info("Mkdocs With Confluence: Run with 'cleanup_orphaned_pages: true' to delete them")

    # Display page status summary in debug/verbose mode
    if self.page_status and (self.config["debug"] or self.config["verbose"]):
        logger.info("\n" + "=" * 80)
        logger.info("PAGE STATUS SUMMARY")
        logger.info("=" * 80)

        # Count statuses
        status_counts: dict[str, int] = {}
        for status in self.page_status.values():
            status_counts[status] = status_counts.get(status, 0) + 1

        # Display summary by status
        for status in ["UPDATE", "NEW PAGE", "NO CHANGE", "WOULD UPDATE (dryrun)", "WOULD CREATE (dryrun)"]:
            if status in status_counts:
                logger.info(f"\n{status}: {status_counts[status]} page(s)")
                pages_with_status: list[str] = [
                    page_name for page_name, s in self.page_status.items() if s == status
                ]
                for page_name in sorted(pages_with_status):
                    logger.info(f"  - {page_name}")

        logger.info("\n" + "=" * 80)
        logger.info(f"TOTAL: {len(self.page_status)} page(s) processed")
        logger.info("=" * 80 + "\n")

get_file_sha1

get_file_sha1(file_path: str) -> str

Calculate SHA1 hash of file.

Parameters:

Name	Type	Description	Default
file_path	str or Path	Path to the file on disk.	required

Returns:

Name	Type	Description
str	str	Hexadecimal SHA1 digest used to tag attachment versions.

Source code in src/mkdocs_to_confluence/plugin.py

def get_file_sha1(self, file_path: str) -> str:
    """Calculate SHA1 hash of file.

    Args:
        file_path (str or Path): Path to the file on disk.

    Returns:
        str: Hexadecimal SHA1 digest used to tag attachment versions.

    """
    hash_sha1 = hashlib.sha1()  # noqa: S324  # nosec B324  # SHA1 for file versioning, not security
    with open(file_path, "rb") as f:
        for chunk in iter(lambda: f.read(4096), b""):
            hash_sha1.update(chunk)
    return hash_sha1.hexdigest()

add_or_update_attachment

add_or_update_attachment(
    page_name: str, filepath: str
) -> None

Add or update attachment on Confluence page.

Parameters:

Name	Type	Description	Default
page_name	str	Title of the page that owns the attachment.	required
filepath	str or Path	File to upload or compare against existing attachments.	required

Source code in src/mkdocs_to_confluence/plugin.py

def add_or_update_attachment(self, page_name: str, filepath: str) -> None:
    """Add or update attachment on Confluence page.

    Args:
        page_name (str): Title of the page that owns the attachment.
        filepath (str or Path): File to upload or compare against existing attachments.

    """
    filename = os.path.basename(filepath)

    page_id = self.find_page_id(page_name)
    if page_id:
        file_hash = self.get_file_sha1(filepath)
        attachment_message = f"MKDocsWithConfluence [v{file_hash}]"
        existing_attachment = self.get_attachment(page_id, filepath)
        if existing_attachment:
            file_hash_regex = re.compile(r"\[v([a-f0-9]{40})]$")
            existing_match = file_hash_regex.search(existing_attachment["version"]["message"])
            if existing_match is not None and existing_match.group(1) == file_hash:
                # Attachment exists and hash matches - skip
                logger.info(f"  * Attachment: {filename} - *NO CHANGE*")
            else:
                # Hash mismatch - update needed
                logger.info(f"  * Attachment: {filename} - *UPDATE*")
                self.update_attachment(page_id, filepath, existing_attachment, attachment_message)
        else:
            # Attachment doesn't exist - create it
            logger.info(f"  * Attachment: {filename} - *NEW*")
            self.create_attachment(page_id, filepath, attachment_message)
    else:
        if self.config["debug"]:
            logger.info("PAGE DOES NOT EXISTS")

get_attachment

get_attachment(
    page_id: str, filepath: str
) -> dict[str, Any] | None

Get existing attachment from Confluence page.

Parameters:

Name	Type	Description	Default
page_id	str	Identifier of the Confluence page.	required
filepath	str or Path	Path whose filename should be matched remotely.	required

Returns:

Type	Description
dict[str, Any] | None	dict or None: Attachment metadata when found, otherwise None.

Source code in src/mkdocs_to_confluence/plugin.py

def get_attachment(self, page_id: str, filepath: str) -> dict[str, Any] | None:
    """Get existing attachment from Confluence page.

    Args:
        page_id (str): Identifier of the Confluence page.
        filepath (str or Path): Path whose filename should be matched remotely.

    Returns:
        dict or None: Attachment metadata when found, otherwise None.

    """
    name = os.path.basename(filepath)
    if self.config["debug"]:
        logger.info(f" * Mkdocs With Confluence: Get Attachment: PAGE ID: {page_id}, FILE: {filepath}")

    url = self.config["host_url"] + "/" + page_id + "/child/attachment"
    headers = {"X-Atlassian-Token": "no-check"}  # no content-type here!
    if self.config["debug"]:
        logger.info(f"URL: {url}")

    r = self._safe_request(
        "get", url, f"getting attachment '{name}'",
        headers=headers, params={"filename": name, "expand": "version"}
    )
    if r is None:
        return None
    with nostdout():
        response_json = r.json()
    if response_json["size"]:
        return response_json["results"][0]
    return None

update_attachment

update_attachment(
    page_id, filepath, existing_attachment, message
)

Update existing attachment on Confluence page.

Parameters:

Name	Type	Description	Default
page_id	str	Identifier of the Confluence page.	required
filepath	str or Path	Local file whose contents replace the attachment.	required
existing_attachment	dict	Metadata returned from get_attachment.	required
message	str	Version comment displayed in Confluence history.	required

Source code in src/mkdocs_to_confluence/plugin.py

def update_attachment(self, page_id, filepath, existing_attachment, message):
    """Update existing attachment on Confluence page.

    Args:
        page_id (str): Identifier of the Confluence page.
        filepath (str or Path): Local file whose contents replace the attachment.
        existing_attachment (dict): Metadata returned from get_attachment.
        message (str): Version comment displayed in Confluence history.

    """
    if self.config["debug"]:
        logger.info(f" * Mkdocs With Confluence: Update Attachment: PAGE ID: {page_id}, FILE: {filepath}")

    url = self.config["host_url"] + "/" + page_id + "/child/attachment/" + existing_attachment["id"] + "/data"
    headers = {"X-Atlassian-Token": "no-check"}  # no content-type here!

    if self.config["debug"]:
        logger.info(f"URL: {url}")

    filename = os.path.basename(filepath)

    # determine content-type
    content_type, encoding = mimetypes.guess_type(filepath)
    if content_type is None:
        content_type = "multipart/form-data"

    if not self.dryrun:
        with open(Path(filepath), "rb") as file_handle:
            files = {"file": (filename, file_handle, content_type), "comment": message}
            r = self._safe_request("post", url, f"updating attachment '{filename}'", headers=headers, files=files)
            if r is None:
                return
            logger.info(r.json())
            if r.status_code == 200:
                logger.info("OK!")
            else:
                logger.error("ERR!")

create_attachment

create_attachment(page_id, filepath, message)

Create new attachment on Confluence page.

Parameters:

Name	Type	Description	Default
page_id	str	Identifier of the Confluence page.	required
filepath	str or Path	Local file to upload as a new attachment.	required
message	str	Version comment displayed in Confluence history.	required

Source code in src/mkdocs_to_confluence/plugin.py

def create_attachment(self, page_id, filepath, message):
    """Create new attachment on Confluence page.

    Args:
        page_id (str): Identifier of the Confluence page.
        filepath (str or Path): Local file to upload as a new attachment.
        message (str): Version comment displayed in Confluence history.

    """
    if self.config["debug"]:
        logger.info(f" * Mkdocs With Confluence: Create Attachment: PAGE ID: {page_id}, FILE: {filepath}")

    url = self.config["host_url"] + "/" + page_id + "/child/attachment"
    headers = {"X-Atlassian-Token": "no-check"}  # no content-type here!

    if self.config["debug"]:
        logger.info(f"URL: {url}")

    filename = os.path.basename(filepath)

    # determine content-type
    content_type, encoding = mimetypes.guess_type(filepath)
    if content_type is None:
        content_type = "multipart/form-data"

    if not self.dryrun:
        with open(filepath, "rb") as file_handle:
            files = {"file": (filename, file_handle, content_type), "comment": message}
            r = self._safe_request("post", url, f"creating attachment '{filename}'", headers=headers, files=files)
            if r is None:
                return
            logger.info(r.json())
            if r.status_code == 200:
                logger.info("OK!")
            else:
                logger.error("ERR!")

find_page_id

find_page_id(page_name: str) -> str | None

Find Confluence page ID by name.

Parameters:

Name	Type	Description	Default
page_name	str	Title of the page to locate.	required

Returns:

Type	Description
str | None	str or None: Page identifier when found, otherwise None.

Source code in src/mkdocs_to_confluence/plugin.py

def find_page_id(self, page_name: str) -> str | None:
    """Find Confluence page ID by name.

    Args:
        page_name (str): Title of the page to locate.

    Returns:
        str or None: Page identifier when found, otherwise None.

    """
    if self.config["debug"]:
        logger.info(f"  * Mkdocs With Confluence: Find Page ID: PAGE NAME: {page_name}")

    name_confl = page_name.replace(" ", "+")
    url = (
        f"{self.config['host_url']}?title={name_confl}"
        f"&spaceKey={self.config['space']}&expand=history"
    )

    if self.config["debug"]:
        logger.info(f"URL: {url}")

    r = self._safe_request("get", url, f"finding page ID for '{page_name}'")
    if r is None:
        return None
    with nostdout():
        response_json = r.json()
    if response_json["results"]:
        if self.config["debug"]:
            logger.info(f"ID: {response_json['results'][0]['id']}")
        return response_json["results"][0]["id"]
    else:
        if self.config["debug"]:
            logger.info("PAGE DOES NOT EXIST")
        return None

get_page_content

get_page_content(page_id: str) -> str | None

Fetch the current content of a page from Confluence.

Parameters:

Name	Type	Description	Default
page_id	str	The Confluence page ID	required

Returns:

Type	Description
str | None	str or None: The page content in storage format, or None if fetch failed

Source code in src/mkdocs_to_confluence/plugin.py

def get_page_content(self, page_id: str) -> str | None:
    """Fetch the current content of a page from Confluence.

    Args:
        page_id: The Confluence page ID

    Returns:
        str or None: The page content in storage format, or None if fetch failed

    """
    if self.config["debug"]:
        logger.info(f"Fetching current content for page ID: {page_id}")
    url = self.config["host_url"] + "/" + page_id + "?expand=body.storage"
    r = self._safe_request("get", url, f"fetching content for page ID '{page_id}'")
    if r is None:
        if self.config["debug"]:
            logger.info("Failed to fetch page content (request returned None)")
        return None
    try:
        with nostdout():
            response_json = r.json()
        content = response_json.get("body", {}).get("storage", {}).get("value")
        if self.config["debug"]:
            logger.info(f"Fetched content length: {len(content) if content else 0} characters")
        return content
    except Exception as e:
        if self.config["debug"]:
            logger.info(f"Error parsing page content: {e}")
        return None

add_page

add_page(
    page_name,
    parent_page_id,
    page_content_in_storage_format,
)

Create new page in Confluence.

Parameters:

Name	Type	Description	Default
page_name	str	Title for the new page.	required
parent_page_id	str	Identifier of the parent page.	required
page_content_in_storage_format	str	Body content in Confluence storage format.	required

Source code in src/mkdocs_to_confluence/plugin.py

def add_page(self, page_name, parent_page_id, page_content_in_storage_format):
    """Create new page in Confluence.

    Args:
        page_name (str): Title for the new page.
        parent_page_id (str): Identifier of the parent page.
        page_content_in_storage_format (str): Body content in Confluence storage format.

    """
    logger.info(f"  * Mkdocs With Confluence: {page_name} - *NEW PAGE*")

    if self.config["debug"]:
        logger.info(f" * Mkdocs With Confluence: Adding Page: PAGE NAME: {page_name}, parent ID: {parent_page_id}")
    url = self.config["host_url"] + "/"
    if self.config["debug"]:
        logger.info(f"URL: {url}")
    headers = {"Content-Type": "application/json"}
    space = self.config["space"]
    data = {
        "type": "page",
        "title": page_name,
        "space": {"key": space},
        "ancestors": [{"id": parent_page_id}],
        "body": {"storage": {"value": page_content_in_storage_format, "representation": "storage"}},
    }
    if self.config["debug"]:
        logger.info(f"DATA: {data}")
    if not self.dryrun:
        r = self._safe_request("post", url, f"creating page '{page_name}'", json=data, headers=headers)
        if r is None:
            return
        if r.status_code == 200:
            if self.config["debug"]:
                logger.info("OK!")
        else:
            if self.config["debug"]:
                logger.error("ERR!")

update_page

update_page(page_name, page_content_in_storage_format)

Update existing page in Confluence.

Parameters:

Name	Type	Description	Default
page_name	str	Title of the page to update.	required
page_content_in_storage_format	str	Body content in Confluence storage format.	required

Source code in src/mkdocs_to_confluence/plugin.py

def update_page(self, page_name, page_content_in_storage_format):
    """Update existing page in Confluence.

    Args:
        page_name (str): Title of the page to update.
        page_content_in_storage_format (str): Body content in Confluence storage format.

    """
    page_id = self.find_page_id(page_name)
    logger.info(f"  * Mkdocs With Confluence: {page_name} - *UPDATE*")
    if self.config["debug"]:
        logger.info(f" * Mkdocs With Confluence: Update PAGE ID: {page_id}, PAGE NAME: {page_name}")
    if page_id:
        page_version = self.find_page_version(page_name)
        if page_version is None:
            logger.error(f"Cannot update page '{page_name}': unable to retrieve version")
            return
        page_version = page_version + 1
        url = self.config["host_url"] + "/" + page_id
        if self.config["debug"]:
            logger.info(f"URL: {url}")
        headers = {"Content-Type": "application/json"}
        space = self.config["space"]
        data = {
            "id": page_id,
            "title": page_name,
            "type": "page",
            "space": {"key": space},
            "body": {"storage": {"value": page_content_in_storage_format, "representation": "storage"}},
            "version": {"number": page_version},
        }

        if not self.dryrun:
            r = self._safe_request("put", url, f"updating page '{page_name}'", json=data, headers=headers)
            if r is None:
                return
            if r.status_code == 200:
                if self.config["debug"]:
                    logger.info("OK!")
            else:
                if self.config["debug"]:
                    logger.error("ERR!")
    else:
        if self.config["debug"]:
            logger.info("PAGE DOES NOT EXIST YET!")

find_page_version

find_page_version(page_name: str) -> int | None

Find current version number of Confluence page.

Parameters:

Name	Type	Description	Default
page_name	str	Title of the page to inspect.	required

Returns:

Type	Description
int | None	int or None: Latest version number, or None when the page is missing.

Source code in src/mkdocs_to_confluence/plugin.py

def find_page_version(self, page_name: str) -> int | None:
    """Find current version number of Confluence page.

    Args:
        page_name (str): Title of the page to inspect.

    Returns:
        int or None: Latest version number, or None when the page is missing.

    """
    if self.config["debug"]:
        logger.info(f"  * Mkdocs With Confluence: Find PAGE VERSION, PAGE NAME: {page_name}")
    name_confl = page_name.replace(" ", "+")
    url = self.config["host_url"] + "?title=" + name_confl + "&spaceKey=" + self.config["space"] + "&expand=version"
    r = self._safe_request("get", url, f"finding page version for '{page_name}'")
    if r is None:
        return None
    with nostdout():
        response_json = r.json()
    if response_json["results"] and len(response_json["results"]) > 0:
        if self.config["debug"]:
            logger.info(f"VERSION: {response_json['results'][0]['version']['number']}")
        return response_json["results"][0]["version"]["number"]
    else:
        if self.config["debug"]:
            logger.info("PAGE DOES NOT EXISTS")
        return None

find_parent_name_of_page

find_parent_name_of_page(name: str) -> str | None

Find parent page name of given Confluence page.

Parameters:

Name	Type	Description	Default
name	str	Title of the page whose parent is requested.	required

Returns:

Type	Description
str | None	str or None: Title of the direct parent page, or None when unavailable.

Source code in src/mkdocs_to_confluence/plugin.py

def find_parent_name_of_page(self, name: str) -> str | None:
    """Find parent page name of given Confluence page.

    Args:
        name (str): Title of the page whose parent is requested.

    Returns:
        str or None: Title of the direct parent page, or None when unavailable.

    """
    if self.config["debug"]:
        logger.info(f"  * Mkdocs With Confluence: Find PARENT OF PAGE, PAGE NAME: {name}")
    idp = self.find_page_id(name)
    if idp is None:
        return None
    url = self.config["host_url"] + "/" + idp + "?expand=ancestors"

    r = self._safe_request("get", url, f"finding parent of page '{name}'")
    if r is None:
        return None
    with nostdout():
        response_json = r.json()
    if response_json and "ancestors" in response_json and len(response_json["ancestors"]) > 0:
        if self.config["debug"]:
            logger.info(f"PARENT NAME: {response_json['ancestors'][-1]['title']}")
        return response_json["ancestors"][-1]["title"]
    else:
        if self.config["debug"]:
            logger.info("PAGE DOES NOT HAVE PARENT")
        return None

get_all_child_pages

get_all_child_pages(parent_id: str) -> list[dict[str, Any]]

Get all child pages recursively under a parent page.

Parameters:

Name	Type	Description	Default
parent_id	str	The ID of the parent page	required

Returns:

Type	Description
list[dict[str, Any]]	List of page dictionaries with 'id' and 'title' keys

Source code in src/mkdocs_to_confluence/plugin.py

def get_all_child_pages(self, parent_id: str) -> list[dict[str, Any]]:
    """Get all child pages recursively under a parent page.

    Args:
        parent_id: The ID of the parent page

    Returns:
        List of page dictionaries with 'id' and 'title' keys

    """
    all_pages: list[dict[str, Any]] = []
    url = self.config["host_url"] + "/" + parent_id + "/child/page"
    params = {"limit": 100}  # Confluence API default limit is 25

    if self.config["debug"]:
        logger.debug(f"Fetching child pages for parent ID: {parent_id}")

    while url:
        r = self._safe_request("get", url, f"fetching child pages for parent ID '{parent_id}'", params=params)
        if r is None:
            break

        with nostdout():
            response_json = r.json()

        if "results" in response_json:
            for page in response_json["results"]:
                page_info = {"id": page["id"], "title": page["title"]}
                all_pages.append(page_info)

                if self.config["debug"]:
                    logger.debug(f"Found child page: {page['title']} (ID: {page['id']})")

                # Recursively get children of this page
                child_pages = self.get_all_child_pages(page["id"])
                all_pages.extend(child_pages)

        # Handle pagination
        if "next" in response_json.get("_links", {}):
            url = self.config["host_url"].rsplit("/content", 1)[0] + response_json["_links"]["next"]
            params = {}  # params are in the URL already
        else:
            break

    return all_pages

add_page_labels

add_page_labels(page_id: str, labels: list[str]) -> None

Add labels to a Confluence page.

Parameters:

Name	Type	Description	Default
page_id	str	The ID of the page	required
labels	list[str]	List of label names to add	required

Source code in src/mkdocs_to_confluence/plugin.py

def add_page_labels(self, page_id: str, labels: list[str]) -> None:
    """Add labels to a Confluence page.

    Args:
        page_id: The ID of the page
        labels: List of label names to add

    """
    if not labels:
        return

    url = self.config["host_url"] + "/" + page_id + "/label"
    headers = {"Content-Type": "application/json"}
    data = [{"name": label, "prefix": "global"} for label in labels]

    if self.config["debug"]:
        logger.debug(f"Adding labels {labels} to page ID: {page_id}")

    if not self.dryrun:
        r = self._safe_request("post", url, f"adding labels to page ID '{page_id}'", json=data, headers=headers)
        if r is not None and self.config["debug"]:
            logger.debug(f"Labels added successfully to page ID: {page_id}")

delete_page

delete_page(page_id: str) -> None

Delete a Confluence page.

Parameters:

Name	Type	Description	Default
page_id	str	The ID of the page to delete	required

Source code in src/mkdocs_to_confluence/plugin.py

def delete_page(self, page_id: str) -> None:
    """Delete a Confluence page.

    Args:
        page_id: The ID of the page to delete

    """
    url = self.config["host_url"] + "/" + page_id

    if self.config["debug"]:
        logger.debug(f"Deleting page ID: {page_id}")

    if not self.dryrun:
        r = self._safe_request("delete", url, f"deleting page ID '{page_id}'")
        if r is not None:
            logger.info(f"Deleted page ID: {page_id}")
        else:
            logger.error(f"Failed to delete page ID: {page_id}")

wait_until

wait_until(
    condition: Callable[[], bool] | bool,
    interval: float = 0.1,
    timeout: int = 10,
    max_retries: int = 3,
) -> bool

Wait until a condition is met, with retry mechanism.

Parameters:

Name	Type	Description	Default
condition	Callable[[], bool] | bool	The condition to wait for (can be a boolean or callable)	required
interval	float	Time between checks in seconds	0.1
timeout	int	Maximum time to wait in seconds	10
max_retries	int	Maximum number of retries if condition is not met	3

Returns:

Type	Description
bool	True if condition is met, False otherwise

Source code in src/mkdocs_to_confluence/plugin.py

def wait_until(
    self, condition: Callable[[], bool] | bool, interval: float = 0.1, timeout: int = 10, max_retries: int = 3
) -> bool:
    """Wait until a condition is met, with retry mechanism.

    Args:
        condition: The condition to wait for (can be a boolean or callable)
        interval: Time between checks in seconds
        timeout: Maximum time to wait in seconds
        max_retries: Maximum number of retries if condition is not met

    Returns:
        True if condition is met, False otherwise

    """
    for retry in range(max_retries):
        start = time.time()
        while time.time() - start < timeout:
            # Evaluate condition - if it's callable, call it; otherwise check truthiness
            result = condition() if callable(condition) else condition
            if result:
                return True
            time.sleep(interval)

        if retry < max_retries - 1:
            logger.info(f"Condition not met, retrying ({retry+1}/{max_retries})...")

    logger.error(f"Condition not met after {max_retries} retries with {timeout}s timeout")
    return False

nostdout

nostdout()

Temporarily silence sys.stdout within the managed context.

Source code in src/mkdocs_to_confluence/plugin.py

@contextlib.contextmanager
def nostdout():
    """Temporarily silence ``sys.stdout`` within the managed context."""
    save_stdout = sys.stdout
    sys.stdout = DummyFile()
    yield
    sys.stdout = save_stdout