Method get_connection

web/pgadmin/utils/driver/psycopg3/__init__.py:194–249

get_connection(...) Returns the connection object for the certain connection-id/database for the specific server, identified by sid. Create a new Connection object (if require). If server id is 0, it will return the database connection object. Param

(
            self, sid, database=None, conn_id=None, auto_reconnect=True,
            **kwargs
    )

        )

    def get_connection(
            self, sid, database=None, conn_id=None, auto_reconnect=True,
            **kwargs
    ):
        """
        get_connection(...)

        Returns the connection object for the certain connection-id/database
        for the specific server, identified by sid. Create a new Connection
        object (if require). If server id is 0, it will return the database
        connection object.

        Parameters:
            sid
            - Server ID
            database
            - Database, on which the connection needs to be made
              If provided none, maintenance_db for the server will be used,
              while generating new Connection object.
            conn_id
            - Identification String for the Connection This will be used by
              certain tools, which will require a dedicated connection for it.
              i.e. Debugger, Query Tool, etc.
            auto_reconnect
            - This parameters define, if we should attempt to reconnect the
              database server automatically, when connection has been lost for
              any reason. Certain tools like Debugger will require a permenant
              connection, and it stops working on disconnection.

        """

        if sid:
            manager = self.connection_manager(sid)

            return manager.connection(database=database, conn_id=conn_id,
                                      auto_reconnect=auto_reconnect)
        else:
            return psycopg.Connection.connect(
                host=kwargs['host'],
                dbname=database,
                user=kwargs['user'],
                password=kwargs[
                    'password'] if 'password' in kwargs else None,
                port=kwargs['port'] if
                kwargs['port'] else None,
                passfile=kwargs['passfile'],
                connect_timeout=kwargs['connect_timeout'] if
                'connect_timeout' in kwargs and
                kwargs['connect_timeout'] else 0,
                sslmode=kwargs['sslmode'],
                sslcert=kwargs['sslcert'],
                sslkey=kwargs['sslkey'],
                sslrootcert=kwargs['sslrootcert'],
                sslcompression=True if kwargs[
                    'sslcompression'] else False
            )

    def release_connection(self, sid, database=None, conn_id=None):