Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def use(self):
""" Run a set of batch commands or start the tool in interactive mode.
"""
if self.env.interactive:
if version_info >= (3,):
self.console.write_help(WELCOME)
else:
self.console.write_help(WELCOME, end=u"")
try:
self.env.connect(NEO4J_URI, NEO4J_USER, NEO4J_PASSWORD)
except ServiceUnavailable:
self.console.write_error("Service Unavailable: Unable to connect to database on %s" % NEO4J_URI)
exit(1)
self.interact()
else:
try:
self.env.connect(NEO4J_URI, NEO4J_USER, NEO4J_PASSWORD)
except ServiceUnavailable:
self.console.write_error("Service Unavailable: Unable to connect to database on %s" % NEO4J_URI)
exit(1)
for arg in self.args[1:]:
self.runsource(arg)
def __init__(self, uri, auth, secure=True, verbose=False):
try:
self.driver = GraphDatabase.driver(uri, auth=auth, encrypted=secure)
except ServiceUnavailable as error:
raise ConsoleError("Could not connect to {} ({})".format(uri, error))
self.uri = uri
self.history = FileHistory(HISTORY_FILE)
self.prompt_args = {
"history": self.history,
"lexer": PygmentsLexer(CypherLexer),
"style": style_from_pygments(VimStyle, {
Token.Prompt: "#ansi{}".format(self.prompt_colour.replace("cyan", "teal")),
Token.TxCounter: "#ansi{} bold".format(self.tx_colour.replace("cyan", "teal")),
})
}
self.lexer = CypherLexer()
self.result_writer = TabularResultWriter()
if verbose:
from .watcher import watch
self.watcher = watch("neo4j.bolt")
def request(self, method, url, fields=None, headers=None, **urlopen_kw):
from neo4j.v1 import ServiceUnavailable
from urllib3.exceptions import MaxRetryError
try:
if self.verified:
return self._http.request(method, url, fields, headers, **urlopen_kw)
else:
with catch_warnings():
simplefilter("ignore")
return self._http.request(method, url, fields, headers, **urlopen_kw)
except MaxRetryError:
raise ServiceUnavailable("Cannot send %s request to <%s>" % (method, url))
if not user:
user = DEFAULT_USER
while True:
graph_service = GraphService(host=host, user=user, password=password, bolt=True)
try:
_ = graph_service.kernel_version
except Unauthorized:
password = getpass("Enter password for user %s: " % user)
except Forbidden:
if graph_service.password_change_required():
self.write("Password expired\n")
new_password = getpass("Enter new password for user %s: " % user)
password = graph_service.change_password(new_password)
else:
raise
except ServiceUnavailable:
self.write("Cannot connect to %s\n" % graph_service.address.host)
raise
else:
self.write("Connected to %s\n" % host)
self.services[host] = graph_service
break
self.graph_service = self.services[host]
self.run = self.graph_service.graph.run
readline.set_completer(SimpleCompleter(self.graph_service, cypher_keywords).complete)
return 0