# SPDX-License-Identifier: GPL-3.0-or-later # # This file is part of Nominatim. (https://nominatim.org) # # Copyright (C) 2023 by the Nominatim developer community. # For a full list of authors see the git log. """ Functions for specialised logging with HTML output. """ from typing import Any, Optional, cast from contextvars import ContextVar import io import sqlalchemy as sa from sqlalchemy.ext.asyncio import AsyncConnection try: from pygments import highlight from pygments.lexers import PythonLexer, PostgresLexer from pygments.formatters import HtmlFormatter CODE_HIGHLIGHT = True except ModuleNotFoundError: CODE_HIGHLIGHT = False class BaseLogger: """ Interface for logging function. The base implementation does nothing. Overwrite the functions in derived classes which implement logging functionality. """ def get_buffer(self) -> str: return '' def function(self, func: str, **kwargs: Any) -> None: """ Start a new debug chapter for the given function and its parameters. """ def section(self, heading: str) -> None: """ Start a new section with the given title. """ def comment(self, text: str) -> None: """ Add a simple comment to the debug output. """ def var_dump(self, heading: str, var: Any) -> None: """ Print the content of the variable to the debug output prefixed by the given heading. """ def sql(self, conn: AsyncConnection, statement: 'sa.Executable') -> None: """ Print the SQL for the given statement. """ class HTMLLogger(BaseLogger): """ Logger that formats messages in HTML. """ def __init__(self): self.buffer = io.StringIO() def get_buffer(self) -> str: return HTML_HEADER + self.buffer.getvalue() + HTML_FOOTER def function(self, func: str, **kwargs: Any) -> None: """ Start a new debug chapter for the given function and its parameters. """ self._write(f"

Debug output for {func}()

\n

Parameters:

") for name, value in kwargs.items(): self._write(f'
{name}
{self._python_var(value)}
') self._write('

') def section(self, heading: str) -> None: """ Start a new section with the given title. """ self._write(f"

{heading}

") def comment(self, text: str) -> None: """ Add a simple comment to the debug output. """ self._write(f"

{text}

") def var_dump(self, heading: str, var: Any) -> None: """ Print the content of the variable to the debug output prefixed by the given heading. """ self._write(f'
{heading}
{self._python_var(var)}') def sql(self, conn: AsyncConnection, statement: 'sa.Executable') -> None: """ Dump the SQL statement to debug output. """ sqlstr = str(cast('sa.ClauseElement', statement) .compile(conn.sync_engine, compile_kwargs={"literal_binds": True})) if CODE_HIGHLIGHT: sqlstr = highlight(sqlstr, PostgresLexer(), HtmlFormatter(nowrap=True, lineseparator='
')) self._write(f'
{sqlstr}
') else: self._write(f'{sqlstr}') def _python_var(self, var: Any) -> str: if CODE_HIGHLIGHT: fmt = highlight(repr(var), PythonLexer(), HtmlFormatter(nowrap=True)) return f'
{fmt}
' return f'{str(var)}' def _write(self, text: str) -> None: """ Add the raw text to the debug output. """ self.buffer.write(text) logger: ContextVar[BaseLogger] = ContextVar('logger', default=BaseLogger()) def set_log_output(fmt: str) -> None: """ Enable collecting debug information. """ if fmt == 'html': logger.set(HTMLLogger()) elif fmt == 'text': logger.set(TextLogger()) else: logger.set(BaseLogger()) def log() -> BaseLogger: """ Return the logger for the current context. """ return logger.get() def get_and_disable() -> str: """ Return the current content of the debug buffer and disable logging. """ buf = logger.get().get_buffer() logger.set(BaseLogger()) return buf HTML_HEADER: str = """ Nominatim - Debug """ HTML_FOOTER: str = ""