fixed PR comments

Author: DvirDukhan

redisgraph_bulk_loader/bulk_insert.py

@@ -70,21 +70,23 @@ async def process_entities(entities):
 @click.option('--max-token-size', '-t', default=500, help='max size of each token in megabytes (default 500, max 512)')
 @click.option('--index', '-i', multiple=True, help='Label:Propery on which to create an index')
 @click.option('--full-text-index', '-f', multiple=True, help='Label:Propery on which to create an full text search index')
-@click.option('--async-requests', '-A', default=3, help='amount of async requests to be executed in parallel' )
+@click.option('--async-requests', '-A', default=3, help='number of async requests to be executed in parallel' )
 async def bulk_insert(graph, host, port, password, user, unix_socket_path, nodes, nodes_with_label, relations, relations_with_type, separator, enforce_schema, skip_invalid_nodes, skip_invalid_edges, escapechar, quote, max_token_count, max_buffer_size, max_token_size, index, full_text_index, async_requests):
-    if sys.version_info.major < 3 or sys.version_info.minor < 6:
-        raise Exception("Python >= 3.6 is required for the RedisGraph bulk loader.")
+    if sys.version_info.major < 3 or sys.version_info.minor < 8:
+        raise Exception("Python >= 3.8 is required for the RedisGraph bulk loader.")
 
     if not (any(nodes) or any(nodes_with_label)):
         raise Exception("At least one node file must be specified.")
+    if async_requests <= 0:
+        raise Exception("The number of async requests must be greater than zero")
 
     start_time = timer()
 
     # If relations are being built, we must store unique node identifiers to later resolve endpoints.
     store_node_identifiers = any(relations) or any(relations_with_type)
 
     # Initialize configurations with command-line arguments
-    config = Config(max_token_count, max_buffer_size, max_token_size, enforce_schema, skip_invalid_nodes, skip_invalid_edges, separator, int(quote), store_node_identifiers, escapechar)
+    config = Config(max_token_count, max_buffer_size, max_token_size, enforce_schema, skip_invalid_nodes, skip_invalid_edges, separator, int(quote), store_node_identifiers, escapechar, async_requests)
 
     # Attempt to connect to Redis server
     try:
@@ -112,7 +114,7 @@ async def bulk_insert(graph, host, port, password, user, unix_socket_path, nodes
         print("Graph with name '%s', could not be created, as Redis key '%s' already exists." % (graph, graph))
         sys.exit(1)
 
-    query_buf = QueryBuffer(graph, client, config, async_requests)
+    query_buf = QueryBuffer(graph, client, config)
 
     # Read the header rows of each input CSV and save its schema.
     labels = parse_schemas(Label, query_buf, nodes, nodes_with_label, config)

redisgraph_bulk_loader/config.py

@@ -1,5 +1,5 @@
 class Config:
-    def __init__(self, max_token_count=1024 * 1023, max_buffer_size=2_048, max_token_size=512, enforce_schema=False, skip_invalid_nodes=False, skip_invalid_edges=False, separator=',', quoting=3, store_node_identifiers=False, escapechar='\\'):
+    def __init__(self, max_token_count=1024 * 1023, max_buffer_size=2_048, max_token_size=512, enforce_schema=False, skip_invalid_nodes=False, skip_invalid_edges=False, separator=',', quoting=3, store_node_identifiers=False, escapechar='\\', async_requests =3 ):
         """Settings for this run of the bulk loader"""
         # Maximum number of tokens per query
         # 1024 * 1024 is the hard-coded Redis maximum. We'll set a slightly lower limit so
@@ -18,6 +18,7 @@ def __init__(self, max_token_count=1024 * 1023, max_buffer_size=2_048, max_token
         self.separator = separator
         self.quoting = quoting
         self.escapechar = None if escapechar.lower() == "none" else escapechar
+        self.async_requests = async_requests
 
         # True if we are building relations as well as nodes
         self.store_node_identifiers = store_node_identifiers

redisgraph_bulk_loader/query_buffer.py

@@ -29,16 +29,16 @@ def send_buffer(self, initial_query):
         return self.client.execute_command("GRAPH.BULK", self.graphname, *args)
 
 class QueryBuffer:
-    def __init__(self, graphname, client, config, async_requests):
+    def __init__(self, graphname, client, config):
 
         self.client = client
         self.graphname = graphname
         self.config = config
-        self.async_requests = async_requests
+        self.async_requests = config.async_requests
 
         # A queue of internal buffers
         self.internal_buffers = list()
-        for i in range(async_requests):
+        for i in range(self.async_requests):
             self.internal_buffers.append(InternalBuffer(graphname, client))
         # Each buffer sent to RedisGraph returns awaitable
         self.awaitables = set()
@@ -60,9 +60,9 @@ def __init__(self, graphname, client, config, async_requests):
     async def send_buffer(self, flush=False):
         # If flush is needed all of the awaitables need to be complete, otherwise at least one is needed.
         return_when_flag = asyncio.ALL_COMPLETED if flush is True else asyncio.FIRST_COMPLETED
-        awaitable = self.current_buffer.send_buffer(self.initial_query)
-        if awaitable is not None:
-            self.awaitables.add(awaitable)
+        coro = self.current_buffer.send_buffer(self.initial_query)
+        if coro is not None:
+            self.awaitables.add(asyncio.create_task(coro))
         # Requests are flushed and awaited when:
         # 1. Flush is needed.
         # 2. Initial query with BEGIN token, to avoid race condition on async RedisGraph servers.