codekingpro/portable-devtools

1"""2libpq debugging tools3 4These functionalities are exposed here for convenience, but are not part of5the public interface and are subject to change at any moment.6 7Suggested usage::8 9    import logging10    import psycopg11    from psycopg import pq12    from psycopg.pq._debug import PGconnDebug13 14    logging.basicConfig(level=logging.INFO, format="%(message)s")15    logger = logging.getLogger("psycopg.debug")16    logger.setLevel(logging.INFO)17 18    assert pq.__impl__ == "python"19    pq.PGconn = PGconnDebug20 21    with psycopg.connect("") as conn:22        conn.pgconn.trace(2)23        conn.pgconn.set_trace_flags(24            pq.Trace.SUPPRESS_TIMESTAMPS | pq.Trace.REGRESS_MODE)25        ...26 27"""28 29# Copyright (C) 2022 The Psycopg Team30 31import inspect32import logging33from typing import Any, Callable, Type, TypeVar, TYPE_CHECKING34from functools import wraps35 36from . import PGconn37from .misc import connection_summary38 39if TYPE_CHECKING:40    from . import abc41 42Func = TypeVar("Func", bound=Callable[..., Any])43 44logger = logging.getLogger("psycopg.debug")45 46 47class PGconnDebug:48    """Wrapper for a PQconn logging all its access."""49 50    _Self = TypeVar("_Self", bound="PGconnDebug")51    _pgconn: "abc.PGconn"52 53    def __init__(self, pgconn: "abc.PGconn"):54        super().__setattr__("_pgconn", pgconn)55 56    def __repr__(self) -> str:57        cls = f"{self.__class__.__module__}.{self.__class__.__qualname__}"58        info = connection_summary(self._pgconn)59        return f"<{cls} {info} at 0x{id(self):x}>"60 61    def __getattr__(self, attr: str) -> Any:62        value = getattr(self._pgconn, attr)63        if callable(value):64            return debugging(value)65        else:66            logger.info("PGconn.%s -> %s", attr, value)67            return value68 69    def __setattr__(self, attr: str, value: Any) -> None:70        setattr(self._pgconn, attr, value)71        logger.info("PGconn.%s <- %s", attr, value)72 73    @classmethod74    def connect(cls: Type[_Self], conninfo: bytes) -> _Self:75        return cls(debugging(PGconn.connect)(conninfo))76 77    @classmethod78    def connect_start(cls: Type[_Self], conninfo: bytes) -> _Self:79        return cls(debugging(PGconn.connect_start)(conninfo))80 81    @classmethod82    def ping(self, conninfo: bytes) -> int:83        return debugging(PGconn.ping)(conninfo)84 85 86def debugging(f: Func) -> Func:87    """Wrap a function in order to log its arguments and return value on call."""88 89    @wraps(f)90    def debugging_(*args: Any, **kwargs: Any) -> Any:91        reprs = []92        for arg in args:93            reprs.append(f"{arg!r}")94        for k, v in kwargs.items():95            reprs.append(f"{k}={v!r}")96 97        logger.info("PGconn.%s(%s)", f.__name__, ", ".join(reprs))98        rv = f(*args, **kwargs)99        # Display the return value only if the function is declared to return100        # something else than None.101        ra = inspect.signature(f).return_annotation102        if ra is not None or rv is not None:103            logger.info("    <- %r", rv)104        return rv105 106    return debugging_  # type: ignore107