(
self,
database=None,
user=None,
password=None,
host=None,
port=None,
dsn=None,
**kwargs,
)
| 193 | return self.__class__(**self._conn_params) |
| 194 | |
| 195 | def connect( |
| 196 | self, |
| 197 | database=None, |
| 198 | user=None, |
| 199 | password=None, |
| 200 | host=None, |
| 201 | port=None, |
| 202 | dsn=None, |
| 203 | **kwargs, |
| 204 | ): |
| 205 | conn_params = self._conn_params.copy() |
| 206 | |
| 207 | new_params = { |
| 208 | "dbname": database, |
| 209 | "user": user, |
| 210 | "password": password, |
| 211 | "host": host, |
| 212 | "port": port, |
| 213 | "dsn": dsn, |
| 214 | } |
| 215 | new_params.update(kwargs) |
| 216 | |
| 217 | if new_params["dsn"]: |
| 218 | new_params = {"dsn": new_params["dsn"], "password": new_params["password"]} |
| 219 | |
| 220 | if new_params["password"]: |
| 221 | new_params["dsn"] = make_conninfo(new_params["dsn"], password=new_params.pop("password")) |
| 222 | |
| 223 | conn_params.update({k: v for k, v in new_params.items() if v}) |
| 224 | |
| 225 | if "dsn" in conn_params: |
| 226 | other_params = {k: v for k, v in conn_params.items() if k != "dsn"} |
| 227 | conn_info = make_conninfo(conn_params["dsn"], **other_params) |
| 228 | else: |
| 229 | conn_info = make_conninfo(**conn_params) |
| 230 | conn = psycopg.connect(conn_info) |
| 231 | conn.cursor_factory = ProtocolSafeCursor |
| 232 | |
| 233 | self._conn_params = conn_params |
| 234 | if self.conn: |
| 235 | self.conn.close() |
| 236 | self.conn = conn |
| 237 | self.conn.autocommit = True |
| 238 | |
| 239 | if self.notify_callback is not None: |
| 240 | self.conn.add_notify_handler(self.notify_callback) |
| 241 | |
| 242 | # When we connect using a DSN, we don't really know what db, |
| 243 | # user, etc. we connected to. Let's read it. |
| 244 | # Note: moved this after setting autocommit because of #664. |
| 245 | dsn_parameters = conn.info.get_parameters() |
| 246 | |
| 247 | if dsn_parameters: |
| 248 | self.dbname = dsn_parameters.get("dbname") |
| 249 | self.user = dsn_parameters.get("user") |
| 250 | self.host = dsn_parameters.get("host") |
| 251 | self.port = dsn_parameters.get("port") |
| 252 | else: |
no test coverage detected