def connection_string_url()

in services/utils/__init__.py [0:0]

• 18 lines of code
• 8 McCabe index (conditional complexity)

    def connection_string_url(self, type=None):
        # postgresql://[user[:password]@][host][:port][/dbname][?param1=value1&...]
        if type is None or type == DBType.WRITER:
            base_url = f'postgresql://{quote(self._user)}:{quote(self._password)}@{self._host}:{self._port}/{self._database_name}'
        elif type == DBType.READER:
            base_url = f'postgresql://{quote(self._user)}:{quote(self._password)}@{self._read_replica_host}:{self._port}/{self._database_name}'
        else:
            raise Exception("Unsupported DBType %s" % type)

        if (self._ssl_mode in ['allow', 'prefer', 'require', 'verify-ca', 'verify-full']):
            ssl_query = f'sslmode={self._ssl_mode}'
            if self._ssl_cert_path is not None:
                ssl_query = f'{ssl_query}&sslcert={self._ssl_cert_path}'
            if self._ssl_key_path is not None:
                ssl_query = f'{ssl_query}&sslkey={self._ssl_key_path}'
            if self._ssl_root_cert_path is not None:
                ssl_query = f'{ssl_query}&sslrootcert={self._ssl_root_cert_path}'
        else:
            ssl_query = f'sslmode=disable'

        return f'{base_url}?{ssl_query}'