host_arg: str = params.get("host", os.environ.get("PGHOST", ""))
hosts_in = host_arg.split(",")
port_arg: str = str(params.get("port", os.environ.get("PGPORT", "")))
- if len((ports_in := port_arg.split(","))) == 1:
+ if len(ports_in := port_arg.split(",")) == 1:
# If only one port is specified, it applies to all the hosts.
ports_in *= len(hosts_in)
if len(ports_in) != len(hosts_in):
self.obj = obj
def __repr__(self) -> str:
- if len((sobj := repr(self.obj))) > 40:
+ if len(sobj := repr(self.obj)) > 40:
sobj = f"{sobj[:35]} ... ({len(sobj)} byteschars)"
return f"{self.__class__.__name__}({sobj})"
return f"<{cls} {info} at 0x{id(self):x}>"
def __getattr__(self, attr: str) -> Any:
- if callable((value := getattr(self._pgconn, attr))):
+ if callable(value := getattr(self._pgconn, attr)):
return debugging(value)
else:
logger.info("PGconn.%s -> %s", attr, value)
import subprocess as sp
libdir = sp.check_output(["pg_config", "--libdir"]).strip().decode()
- if not os.path.exists((libname := os.path.join(libdir, "libpq.dylib"))):
+ if not os.path.exists(libname := os.path.join(libdir, "libpq.dylib")):
libname = None
except Exception as ex:
logger.debug("couldn't use pg_config to find libpq: %s", ex)
# before upgrading the ConnStatus enum.
status = f"status={pgconn.status} (unkndown)"
- if sparts := " ".join(("%s=%s" % part for part in parts)):
+ if sparts := " ".join("%s=%s" % part for part in parts):
sparts = f" ({sparts})"
return f"[{status}]{sparts}"
self.dumps = dumps
def __repr__(self) -> str:
- if len((sobj := repr(self.obj))) > 40:
+ if len(sobj := repr(self.obj)) > 40:
sobj = f"{sobj[:35]} ... ({len(sobj)} chars)"
return f"{self.__class__.__name__}({sobj})"
try:
cur.execute(self._insert_field_stmt(j), (val,))
except psycopg.DatabaseError as e:
- if len((r := repr(val))) > 200:
+ if len(r := repr(val)) > 200:
r = f"{r[:200]}... ({len(r)} chars)"
raise Exception(
f"value {r!r} at record {i} column0 {j} failed insert: {e}"
try:
await acur.execute(self._insert_field_stmt(j), (val,))
except psycopg.DatabaseError as e:
- if len((r := repr(val))) > 200:
+ if len(r := repr(val)) > 200:
r = f"{r[:200]}... ({len(r)} chars)"
raise Exception(
f"value {r!r} at record {i} column0 {j} failed insert: {e}"
def main() -> None:
opt = parse_cmdline()
- if CrdbConnection.is_crdb((conn := psycopg.connect(opt.dsn, autocommit=True))):
+ if CrdbConnection.is_crdb(conn := psycopg.connect(opt.dsn, autocommit=True)):
conn = CrdbConnection.connect(opt.dsn, autocommit=True)
update_crdb_python_oids(conn)
else: