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