how to connect neo4j auradb behind firewall proxy with python api

[furyNerd]

hi, i'm learning to use neo4j auradb just recently, i tried the example of the docs:

https://neo4j.com/docs/aura/auradb/connecting-applications/python/

but got errors:

Unable to retrieve routing information
Transaction failed and will be retried in 1.0940946540965086s (Unable to retrieve routing information)
...
...
...
neo4j.exceptions.ServiceUnavailable: Unable to retrieve routing information

i think it the issue about my internet connection is behind a firewall of my organization and need a proxy http://username:password@ip:port, by this proxy config in browser, i can use the db on web online.

but how can i config it with python api? (running .py on my local pc)

i have tried set the proxy in python script like:

    import os
    os.environ["http_proxy"]  = "http://username:password@ip:port"
    os.environ["https_proxy"] = "http://username:password@ip:port"
    os.environ["ALL_PROXY"]   = "http://username:password@ip:port"
    os.environ["HTTP_PROXY"]  = "http://username:password@ip:port"
    os.environ["HTTPS_PROXY"] = "http://username:password@ip:port"

looks no help...

and btw, gpt told me something like

# Set up the configuration for the proxy
config = {
    'proxies': {
        'http': 'http://your-proxy-url:port',
        'https': 'https://your-proxy-url:port'
    }
}

# Set up the driver with the proxy configuration
driver = GraphDatabase.driver("bolt://your-neo4j-url:7687", auth=basic_auth("your-neo4j-username", "your-neo4j-password"), **config)

looks like total nonescense... i searched the docs and found nothing like proxy configs

thanks for any help

[robsdedude]

Currently, there is no config option for this. Ideally, you'd talk to the IT department of your organization (or who else is in charge of the proxy) and ask them to configure a firewall exception for your database.

For just playing around, here's a quick and dirty hack.⚠️ It's not at all a sustainable solution and putting it into production is a horrible idea. So please take it for what it is.

import neo4j

URI = "neo4j+s://<YOUR AURA URL>"
AUTH = ("neo4j", "<PASSWORD>")


# requires `pip install python-socks`
def apply_dirty_hack():
    """Make the driver use python-socks to connect through a proxy.

    .. warning::
        This is a dirty hack as the name suggest.
        It's written to work with driver version 5.6.0.
        There's no guarantee that it will work with any other version.

        Furthermore, this hack is limited to the synchronous driver only.
        Any async driver api will try to connect to the database directly.
    """

    from socket import (
        SO_KEEPALIVE,
        SOL_SOCKET,
        timeout as SocketTimeout
    )

    from neo4j.exceptions import ServiceUnavailable
    # digging deep into the driver internals
    from neo4j._async_compat.network._bolt_socket import (
        BoltSocket,
        log,
    )
    from python_socks.sync import Proxy


    # copied from neo4j._async_compat.network._bolt_socket
    # and adjusted to use python-socks
    def _connect(cls, resolved_address, timeout, keep_alive):
        s = None

        try:
            # changed: use python-socks to connect through a proxy
            proxy = Proxy.from_url("http://user:passwort@127.0.0.1:1080")
            s = proxy.connect(resolved_address.host, resolved_address.port)
            # end of change
            t = s.gettimeout()
            if timeout:
                s.settimeout(timeout)
            log.debug("[#0000]  C: <OPEN> %s", resolved_address)
            s.connect(resolved_address)
            s.settimeout(t)
            keep_alive = 1 if keep_alive else 0
            s.setsockopt(SOL_SOCKET, SO_KEEPALIVE, keep_alive)
            return s
        except SocketTimeout:
            log.debug("[#0000]  S: <TIMEOUT> %s", resolved_address)
            log.debug("[#0000]  C: <CLOSE> %s", resolved_address)
            cls.close_socket(s)
            raise ServiceUnavailable(
                "Timed out trying to establish connection to {!r}".format(
                    resolved_address))
        except Exception as error:
            log.debug("[#0000]  S: <ERROR> %s %s", type(error).__name__,
                      " ".join(map(repr, error.args)))
            log.debug("[#0000]  C: <CLOSE> %s", resolved_address)
            cls.close_socket(s)
            if isinstance(error, OSError):
                raise ServiceUnavailable(
                    "Failed to establish connection to {!r} (reason {})"
                    .format(resolved_address, error)
                ) from error
            raise

    # fake it till you make it
    # override the driver's connect method with our own
    BoltSocket._connect = classmethod(_connect)


def main():
    apply_dirty_hack()
    # ... use the driver as normal, e.g.:
    with neo4j.GraphDatabase.driver(URI, auth=AUTH) as driver:
        print(driver.execute_query("RETURN 1 AS n"))


if __name__ == "__main__":
    main()

Sadly, I can't make any promises when or if this will ever become a feature supported by the driver. There has been little demand so far and other features have higher priority right now.