@property
def port(self) -> int:
"""The port of the active connection. See :pq:`PQport()`."""
- if sport := self._get_pgconn_attr("port"):
- return int(sport)
- return self._get_compiled_port()
+ if port := self._get_pgconn_attr("port"):
+ return int(port)
- def _get_compiled_port(sef) -> int:
# As per docs: an empty string means the build default, not e.g.
# something configured by PGPORT
# https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-PORT
- for info in pq.Conninfo().get_defaults():
- if info.keyword == b"port":
- if info.compiled:
- return int(info.compiled)
- break
+ elif port := pq.misc.get_compiled_port():
+ return int(port)
raise e.InternalError("couldn't find the connection port")
# Put together the (CONNECTION)
if not pgconn.host.startswith(b"/"):
parts.append(("host", pgconn.host.decode()))
- if pgconn.port != b"5432":
- parts.append(("port", pgconn.port.decode()))
+ if (port := pgconn.port.decode() or get_compiled_port()) != "5432":
+ parts.append(("port", port))
if pgconn.user != pgconn.db:
parts.append(("user", pgconn.user.decode()))
parts.append(("database", pgconn.db.decode()))
return f"{major}.{patch}"
else:
return f"{major}.{minor}.{patch}"
+
+
+@cache
+def get_compiled_port() -> str:
+ """Return the default port compiled with the libpq."""
+
+ from psycopg._conninfo_utils import get_param_def
+
+ info = get_param_def("port")
+ return info.compiled if info and info.compiled else ""
# assume 5432 is the compiled value
assert conn.info.port == 5432
+ assert "port=" not in repr(conn)
+
def test_get_params(conn, dsn):
info = conn.info.get_parameters()