markdown_exec ¤

def add_source(
    *,
    source: str,
    location: str,
    output: str,
    language: str,
    tabs: tuple[str, str],
    result: str = "",
    **extra: str,
) -> str:
    """Add source code block to the output.

    Parameters:
        source: The source code block.
        location: Where to add the source (above, below, tabbed-left, tabbed-right, console).
        output: The current output.
        language: The code language.
        tabs: Tabs titles (if used).
        result: Syntax to use when concatenating source and result with "console" location.
        **extra: Extra options added back to source code block.

    Raises:
        ValueError: When the given location is not supported.

    Returns:
        The updated output.
    """
    source = _hide_lines(source)
    if location == "console":
        return code_block(result or language, source + "\n" + output, **extra)

    source_block = code_block(language, source, **extra)
    if location == "above":
        return source_block + "\n\n" + output
    if location == "below":
        return output + "\n\n" + source_block
    if location == "material-block":
        return source_block + f'\n\n<div class="result" markdown="1" >\n\n{output}\n\n</div>'

    source_tab_title, result_tab_title = tabs
    if location == "tabbed-left":
        return tabbed((source_tab_title, source_block), (result_tab_title, output))
    if location == "tabbed-right":
        return tabbed((result_tab_title, output), (source_tab_title, source_block))

    raise ValueError(f"unsupported location for sources: {location}")

def base_format(
    *,
    language: str,
    run: Callable,
    code: str,
    md: Markdown,
    html: bool = False,
    source: str = "",
    result: str = "",
    tabs: tuple[str, str] = default_tabs,
    id: str = "",  # noqa: A002
    id_prefix: str | None = None,
    returncode: int = 0,
    transform_source: Callable[[str], tuple[str, str]] | None = None,
    session: str | None = None,
    update_toc: bool = True,
    workdir: str | None = None,
    width: int | None = None,
    **options: Any,
) -> Markup:
    """Execute code and return HTML.

    Parameters:
        language: The code language.
        run: Function that runs code and returns output.
        code: The code to execute.
        md: The Markdown instance.
        html: Whether to inject output as HTML directly, without rendering.
        source: Whether to show source as well, and where.
        result: If provided, use as language to format result in a code block.
        tabs: Titles of tabs (if used).
        id: An optional ID for the code block (useful when warning about errors).
        id_prefix: A string used to prefix HTML ids in the generated HTML.
        returncode: The expected exit code.
        transform_source: An optional callable that returns transformed versions of the source.
            The input source is the one that is ran, the output source is the one that is
            rendered (when the source option is enabled).
        session: A session name, to persist state between executed code blocks.
        update_toc: Whether to include generated headings
            into the Markdown table of contents (toc extension).
        workdir: The working directory to use for the execution.
        **options: Additional options passed from the formatter.

    Returns:
        HTML contents.
    """
    markdown = MarkdownConverter(md, update_toc=update_toc)
    extra = options.get("extra", {})

    if transform_source:
        source_input, source_output = transform_source(code)
    else:
        source_input = code
        source_output = code

    try:
        with working_directory(workdir), console_width(width):
            output = run(source_input, returncode=returncode, session=session, id=id, **extra)
    except ExecutionError as error:
        identifier = id or extra.get("title", "")
        identifier = identifier and f"'{identifier}' "
        exit_message = "errors" if error.returncode is None else f"unexpected code {error.returncode}"
        log_message = (
            f"Execution of {language} code block {identifier}exited with {exit_message}\n\n"
            f"Code block is:\n\n{_format_log_details(source_input)}\n\n"
            f"Output is:\n\n{_format_log_details(str(error), strip_fences=True)}\n"
        )
        _logger.warning(log_message)
        return markdown.convert(str(error))

    if not output and not source:
        return Markup()

    if html:
        if source:
            placeholder = f'<div class="{uuid4()}"></div>'
            wrapped_output = add_source(
                source=source_output,
                location=source,
                output=placeholder,
                language=language,
                tabs=tabs,
                **extra,
            )
            return markdown.convert(wrapped_output, stash={placeholder: output})
        return Markup(output)  # noqa: S704

    wrapped_output = output
    if result and source != "console":
        wrapped_output = code_block(result, output)
    if source:
        wrapped_output = add_source(
            source=source_output,
            location=source,
            output=wrapped_output,
            language=language,
            tabs=tabs,
            result=result,
            **extra,
        )
    prefix = id_prefix if id_prefix is not None else (f"{id}-" if id else None)
    return markdown.convert(wrapped_output, id_prefix=prefix)

def code_block(language: str, code: str, **options: str) -> str:
    """Format code as a code block.

    Parameters:
        language: The code block language.
        code: The source code to format.
        **options: Additional options passed from the source, to add back to the generated code block.

    Returns:
        The formatted code block.
    """
    opts = " ".join(f'{opt_name}="{opt_value}"' for opt_name, opt_value in options.items())
    return f"````````{language} {opts}\n{code}\n````````"

@contextmanager
def console_width(width: int | None = None) -> Iterator[None]:
    """Set the console width for the duration of the context.

    The console width is set using the `COLUMNS` environment variable.

    Parameters:
        width: The width to set the console to.
    """
    if width:
        old_width = os.environ.get("COLUMNS", None)
        os.environ["COLUMNS"] = str(width)
        try:
            yield
        finally:
            if old_width is None:
                del os.environ["COLUMNS"]
            else:
                os.environ["COLUMNS"] = old_width
    else:
        yield

def formatter(
    source: str,
    language: str,
    css_class: str,  # noqa: ARG001
    options: dict[str, Any],
    md: Markdown,
    classes: list[str] | None = None,  # noqa: ARG001
    id_value: str = "",  # noqa: ARG001
    attrs: dict[str, Any] | None = None,  # noqa: ARG001
    **kwargs: Any,  # noqa: ARG001
) -> str:
    """Execute code and return HTML.

    Parameters:
        source: The code to execute.
        language: The code language, like python or bash.
        css_class: The CSS class to add to the HTML element.
        options: The container for options.
        attrs: The container for attrs:
        md: The Markdown instance.
        classes: Additional CSS classes.
        id_value: An optional HTML id.
        attrs: Additional attributes
        **kwargs: Additional arguments passed to SuperFences default formatters.

    Returns:
        HTML contents.
    """
    fmt = formatters.get(language, lambda source, **kwargs: source)
    _logger.debug("Executing %s code block with options: %s", language, options)
    return fmt(code=source, md=md, **options)  # type: ignore[operator]

def get_logger(name: str) -> _Logger:
    """Create and return a new logger instance.

    Parameters:
        name: The logger name.

    Returns:
        The logger.
    """
    return _Logger.get(name)

def patch_loggers(get_logger_func: Callable[[str], Any]) -> None:
    """Patch loggers.

    We provide the `patch_loggers`function so dependant libraries
    can patch loggers as they see fit.

    For example, to fit in the MkDocs logging configuration
    and prefix each log message with the module name:

    ```python
    import logging
    from markdown_exec.logger import patch_loggers


    class LoggerAdapter(logging.LoggerAdapter):
        def __init__(self, prefix, logger):
            super().__init__(logger, {})
            self.prefix = prefix

        def process(self, msg, kwargs):
            return f"{self.prefix}: {msg}", kwargs


    def get_logger(name):
        logger = logging.getLogger(f"mkdocs.plugins.{name}")
        return LoggerAdapter(name.split(".", 1)[0], logger)


    patch_loggers(get_logger)
    ```

    Parameters:
        get_logger_func: A function accepting a name as parameter and returning a logger.
    """
    _Logger._patch_loggers(get_logger_func)

def tabbed(*tabs: tuple[str, str]) -> str:
    """Format tabs using `pymdownx.tabbed` extension.

    Parameters:
        *tabs: Tuples of strings: title and text.

    Returns:
        The formatted tabs.
    """
    parts = []
    for title, text in tabs:
        title = title.replace(r"\|", "|").strip()  # noqa: PLW2901
        parts.append(f'=== "{title}"')
        parts.append(indent(text, prefix=" " * 4))
        parts.append("")
    return "\n".join(parts)

def validator(
    language: str,
    inputs: dict[str, str],
    options: dict[str, Any],
    attrs: dict[str, Any],  # noqa: ARG001
    md: Markdown,  # noqa: ARG001
) -> bool:
    """Validate code blocks inputs.

    Parameters:
        language: The code language, like python or bash.
        inputs: The code block inputs, to be sorted into options and attrs.
        options: The container for options.
        attrs: The container for attrs:
        md: The Markdown instance.

    Returns:
        Success or not.
    """
    exec_value = language in MARKDOWN_EXEC_AUTO or _to_bool(inputs.pop("exec", "no"))
    if language not in {"tree", "pyodide"} and not exec_value:
        return False
    id_value = inputs.pop("id", "")
    id_prefix_value = inputs.pop("idprefix", None)
    html_value = _to_bool(inputs.pop("html", "no"))
    source_value = inputs.pop("source", "")
    result_value = inputs.pop("result", "")
    returncode_value = int(inputs.pop("returncode", "0"))
    session_value = inputs.pop("session", "")
    update_toc_value = _to_bool(inputs.pop("updatetoc", "yes"))
    tabs_value = inputs.pop("tabs", "|".join(default_tabs))
    tabs = tuple(_tabs_re.split(tabs_value, maxsplit=1))
    workdir_value = inputs.pop("workdir", None)
    width_value = int(inputs.pop("width", "0"))
    options["id"] = id_value
    options["id_prefix"] = id_prefix_value
    options["html"] = html_value
    options["source"] = source_value
    options["result"] = result_value
    options["returncode"] = returncode_value
    options["session"] = session_value
    options["update_toc"] = update_toc_value
    options["tabs"] = tabs
    options["workdir"] = workdir_value
    options["width"] = width_value
    options["extra"] = inputs
    return True

@contextmanager
def working_directory(path: str | None = None) -> Iterator[None]:
    """Change the working directory for the duration of the context.

    Parameters:
        path: The path to change the working directory to.
    """
    if path:
        old_cwd = os.getcwd()
        os.chdir(path)
        try:
            yield
        finally:
            os.chdir(old_cwd)
    else:
        yield