in src/graph_notebook/configuration/generate_config.py [0:0]
def __init__(self, traversal_source: str = '', username: str = '', password: str = '',
message_serializer: str = '', connection_protocol: str = '',
include_protocol: bool = False, neptune_service: str = ''):
"""
:param traversal_source: used to specify the traversal source for a Gremlin traversal, in the case that we are
connected to an endpoint that can access multiple graphs.
:param username: used to specify a username for authenticating to Gremlin Server, if the endpoint supports it.
:param password: used to specify a password for authenticating to Gremlin Server, if the endpoint supports it.
:param message_serializer: used to specify a serializer for encoding the data to and from Gremlin Server.
:param connection_protocol: used to specify a protocol for the Gremlin connection.
:param include_protocol: used to specify whether to include connection_protocol in the Gremlin config.
"""
if traversal_source == '':
traversal_source = DEFAULT_GREMLIN_TRAVERSAL_SOURCE
invalid_serializer_input = False
if message_serializer != '':
message_serializer, invalid_serializer_input = normalize_serializer_class_name(message_serializer)
if include_protocol:
# Neptune endpoint
invalid_protocol_input = False
if connection_protocol != '':
connection_protocol, invalid_protocol_input = normalize_protocol_name(connection_protocol)
if neptune_service == NEPTUNE_ANALYTICS_SERVICE_NAME:
if connection_protocol != DEFAULT_HTTP_PROTOCOL:
if invalid_protocol_input:
print(f"Invalid connection protocol specified, you must use {DEFAULT_HTTP_PROTOCOL}. ")
elif connection_protocol == DEFAULT_WS_PROTOCOL:
print(f"Enforcing HTTP protocol.")
connection_protocol = DEFAULT_HTTP_PROTOCOL
# temporary restriction until GraphSON-typed and GraphBinary results are supported
if message_serializer not in NEPTUNE_GREMLIN_SERIALIZERS_HTTP_NEXT:
if message_serializer not in GREMLIN_SERIALIZERS_ALL:
if invalid_serializer_input:
print(f"Invalid serializer specified, defaulting to {DEFAULT_GREMLIN_HTTP_SERIALIZER_NEXT}. "
f"Valid serializers: {NEPTUNE_GREMLIN_SERIALIZERS_HTTP_NEXT}")
else:
print(f"{message_serializer} is not currently supported for HTTP connections, "
f"defaulting to {DEFAULT_GREMLIN_HTTP_SERIALIZER_NEXT}. "
f"Please use one of: {NEPTUNE_GREMLIN_SERIALIZERS_HTTP_NEXT}")
message_serializer = DEFAULT_GREMLIN_HTTP_SERIALIZER_NEXT
else:
if connection_protocol not in [DEFAULT_WS_PROTOCOL, DEFAULT_HTTP_PROTOCOL]:
if invalid_protocol_input:
print(f"Invalid connection protocol specified, defaulting to {DEFAULT_WS_PROTOCOL}. "
f"Valid protocols: [websockets, http].")
connection_protocol = DEFAULT_WS_PROTOCOL
if connection_protocol == DEFAULT_HTTP_PROTOCOL:
# temporary restriction until GraphSON-typed and GraphBinary results are supported
if message_serializer not in NEPTUNE_GREMLIN_SERIALIZERS_HTTP:
if message_serializer not in GREMLIN_SERIALIZERS_ALL:
if invalid_serializer_input:
print(f"Invalid serializer specified, defaulting to {DEFAULT_GREMLIN_HTTP_SERIALIZER}. "
f"Valid serializers: {NEPTUNE_GREMLIN_SERIALIZERS_HTTP}")
else:
print(f"{message_serializer} is not currently supported for HTTP connections, "
f"defaulting to {DEFAULT_GREMLIN_HTTP_SERIALIZER}. "
f"Please use one of: {NEPTUNE_GREMLIN_SERIALIZERS_HTTP}")
message_serializer = DEFAULT_GREMLIN_HTTP_SERIALIZER
else:
if message_serializer not in GREMLIN_SERIALIZERS_WS:
if invalid_serializer_input:
print(f"Invalid serializer specified, defaulting to {DEFAULT_GREMLIN_WS_SERIALIZER}. "
f"Valid serializers: {GREMLIN_SERIALIZERS_WS}")
elif message_serializer != '':
print(f"{message_serializer} is not currently supported by Gremlin Python driver, "
f"defaulting to {DEFAULT_GREMLIN_WS_SERIALIZER}. "
f"Valid serializers: {GREMLIN_SERIALIZERS_WS}")
message_serializer = DEFAULT_GREMLIN_WS_SERIALIZER
self.connection_protocol = connection_protocol
else:
# Non-Neptune database - check and set valid WebSockets serializer if invalid/empty
if message_serializer not in GREMLIN_SERIALIZERS_WS:
message_serializer = DEFAULT_GREMLIN_WS_SERIALIZER
if invalid_serializer_input:
print(f'Invalid Gremlin serializer specified, defaulting to {DEFAULT_GREMLIN_WS_SERIALIZER}. '
f'Valid serializers: {GREMLIN_SERIALIZERS_WS}.')
self.traversal_source = traversal_source
self.username = username
self.password = password
self.message_serializer = message_serializer