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.simple_log = False
    self.flen = 1
    self.session = requests.Session()
    self.page_attachments = {}
    self.dryrun = False
    self.exporter = None
    self.synced_pages: set[str] = set()

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_post_template

on_post_template(output_content, template_name, config)

Configure logging mode based on verbosity settings.

Parameters:

Name	Type	Description	Default
output_content	str	Rendered output from MkDocs.	required
template_name	str	Name of the template used for rendering.	required
config	Config	Active MkDocs configuration.	required

Source code in src/mkdocs_to_confluence/plugin.py

def on_post_template(self, output_content, template_name, config):
    """Configure logging mode based on verbosity settings.

    Args:
        output_content (str): Rendered output from MkDocs.
        template_name (str): Name of the template used for rendering.
        config (mkdocs.config.base.Config): Active MkDocs configuration.

    """
    if self.config["verbose"] is False and self.config["debug"] is False:
        self.simple_log = True
        logger.info("Mkdocs With Confluence: Start exporting markdown pages... (simple logging)")
    else:
        self.simple_log = False

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.

    """
    # Load environment variables from .env file
    load_dotenv()

    # Update config with environment variables
    if not self.config.get('host_url') and environ.get('CONFLUENCE_HOST_URL'):
        self.config['host_url'] = environ.get('CONFLUENCE_HOST_URL')
    if not self.config.get('space') and environ.get('SPACE_KEY'):
        self.config['space'] = environ.get('SPACE_KEY')
    if not self.config.get('parent_page_name') and environ.get('PARENT_PAGE_NAME'):
        self.config['parent_page_name'] = environ.get('PARENT_PAGE_NAME')
    if not self.config.get('api_token') and environ.get('CONFLUENCE_API_TOKEN'):
        self.config['api_token'] = environ.get('CONFLUENCE_API_TOKEN')
        # Use bearer authentication when API token is provided
        self.config['auth_type'] = 'bearer'
    if not self.config.get('username') and environ.get('JIRA_USERNAME'):
        self.config['username'] = environ.get('JIRA_USERNAME')
    if not self.config.get('password') and environ.get('JIRA_PASSWORD'):
        self.config['password'] = environ.get('JIRA_PASSWORD')

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

    # Always show configuration status for troubleshooting (without exposing credentials)
    logger.info("Configuration check:")
    logger.info(f"  - host_url configured: {self.config.get('host_url')}")
    logger.info(f"  - space configured: {self.config.get('space')}")
    logger.info(f"  - parent_page_name configured name: {self.config.get('parent_page_name')}")
    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')}")

    # Always set dryrun regardless of enabled status
    if self.config["dryrun"]:
        logger.warning("Mkdocs With Confluence - DRYRUN MODE turned ON")
        self.dryrun = True
        # Initialize exporter for dry-run mode
        export_dir = Path(self.config["export_dir"])
        self.exporter = ConfluenceExporter(export_dir)
        logger.info(f"Mkdocs With Confluence: Exporting to {export_dir}")
    else:
        self.dryrun = 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("Using OAuth Bearer token authentication")
        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 token == "":
                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 == "":
            logger.warning("Password is not configured. Check plugin configuration or environment variables.")
        self.session.auth = (username, password)

    if self.enabled:
        if self.simple_log is True:
            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.dryrun and self.exporter:
                # Add page to exporter queue for dry-run export
                # For dry-run, pass direct parent (last in chain) if it's not the root
                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 mode: sync to Confluence (create or update)
                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.

    """
    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}"):
            self.add_or_update_attachment(page.title, 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 dry-run export
    if self.dryrun and 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

    # Handle orphaned pages detection and cleanup (only when enabled and not in dryrun)
    if not self.enabled or self.dryrun:
        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
    if self.config.get("cleanup_orphaned_pages", False):
        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:
        logger.info("Mkdocs With Confluence: Run with 'cleanup_orphaned_pages: true' to delete them")

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]

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