def __init__()

in ossdbtoolsservice/driver/types/psycopg_driver.py [0:0]

• 24 lines of code
• 13 McCabe index (conditional complexity)

    def __init__(self, conn_params: {}, config: Optional[Configuration] = None):
        """
        Creates a new connection wrapper. Parses version string
        :param conn_params: connection parameters dict
        :param config: optional Configuration object with pgsql connection config
        """
        # If options contains azureSecurityToken, then just copy it over to password, which is how it is
        # passed to PostgreSQL.
        if 'azureAccountToken' in conn_params:
            conn_params['password'] = conn_params['azureAccountToken']

        # Map the connection options to their psycopg2-specific options
        self._connection_options = connection_options = {PG_CONNECTION_OPTION_KEY_MAP.get(option, option): value for option, value in conn_params.items()
                                                         if option in PG_CONNECTION_PARAM_KEYWORDS}

        # Use the default database if one was not provided
        if 'dbname' not in connection_options or not connection_options['dbname']:
            if config:
                connection_options['dbname'] = config.pgsql.default_database
            else:
                connection_options['dbname'] = constants.DEFAULT_DB[constants.PG_PROVIDER_NAME]

        # Use the default port number if one was not provided
        if 'port' not in connection_options or not connection_options['port']:
            connection_options['port'] = constants.DEFAULT_PORT[constants.PG_PROVIDER_NAME]

        # Pass connection parameters as keyword arguments to the connection by unpacking the connection_options dict
        self._conn = psycopg2.connect(**connection_options)

        # Set autocommit mode so that users have control over transactions
        self._conn.autocommit = True

        # Get the DSN parameters for the connection as a dict
        self._dsn_parameters = self._conn.get_dsn_parameters()

        # Find the class of the database error this driver throws
        self._database_error = psycopg2.DatabaseError

        # Calculate the server version
        version_string = str(self._conn.server_version)
        self._version: Tuple[int, int, int] = (
            int(version_string[:-4]),
            int(version_string[-4:-2]),
            int(version_string[-2:])
        )

        # Setting the provider for this connection
        self._provider_name = constants.PG_PROVIDER_NAME
        self._server_type = "PostgreSQL"