Function _parse_connect_dsn_and_args

asyncpg/connect_utils.py:276–859

(*, dsn, host, port, user,
                                password, passfile, database, ssl,
                                service, servicefile,
                                direct_tls, server_settings,
                                target_session_attrs, krbsrvname, gsslib)

def _parse_connect_dsn_and_args(*, dsn, host, port, user,
                                password, passfile, database, ssl,
                                service, servicefile,
                                direct_tls, server_settings,
                                target_session_attrs, krbsrvname, gsslib):
    # `auth_hosts` is the version of host information for the purposes
    # of reading the pgpass file.
    auth_hosts = None
    sslcert = sslkey = sslrootcert = sslcrl = sslpassword = None
    ssl_min_protocol_version = ssl_max_protocol_version = None
    sslnegotiation = None

    if dsn:
        parsed = urllib.parse.urlparse(dsn)

        query = None
        if parsed.query:
            query = urllib.parse.parse_qs(parsed.query, strict_parsing=True)
            for key, val in query.items():
                if isinstance(val, list):
                    query[key] = val[-1]

            if 'service' in query:
                val = query.pop('service')
                if not service and val:
                    service = val

        connection_service_file = servicefile

        if connection_service_file is None:
            connection_service_file = os.getenv('PGSERVICEFILE')

        if connection_service_file is None:
            homedir = compat.get_pg_home_directory()
            if homedir:
                connection_service_file = homedir / PG_SERVICEFILE
            else:
                connection_service_file = None
        else:
            connection_service_file = pathlib.Path(connection_service_file)

        if parsed.scheme not in {'postgresql', 'postgres'}:
            raise exceptions.ClientConfigurationError(
                'invalid DSN: scheme is expected to be either '
                '"postgresql" or "postgres", got {!r}'.format(parsed.scheme))

        if parsed.netloc:
            if '@' in parsed.netloc:
                dsn_auth, _, dsn_hostspec = parsed.netloc.partition('@')
            else:
                dsn_hostspec = parsed.netloc
                dsn_auth = ''
        else:
            dsn_auth = dsn_hostspec = ''

        if dsn_auth:
            dsn_user, _, dsn_password = dsn_auth.partition(':')
        else: