Skip to content

fix(loader): always close CSV file handle on exception via try/finally #75

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
gkorland wants to merge 7 commits into
base: master
Choose a base branch
from
Open

fix(loader): always close CSV file handle on exception via try/finally #75

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
1d43943
fix(loader): close CSV file handle on exception via try/finally
gkorland Apr 26, 2026
caaf535
Merge remote-tracking branch 'origin/master' into fix/csv-file-handle…
Copilot Apr 26, 2026
49febe5
Merge branch 'master' into fix/csv-file-handle-leak
gkorland Apr 29, 2026
e421728
fix(loader): chain SchemaError with 'from e' and guard infile.close()
invalid-email-address Apr 29, 2026
238c255
Potential fix for pull request finding 'Empty except'
gkorland Apr 29, 2026
490bce2
Potential fix for pull request finding 'Empty except'
gkorland Apr 29, 2026
5ce01bf
Merge branch 'master' into fix/csv-file-handle-leak
gkorland Apr 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
91 changes: 48 additions & 43 deletions falkordb_bulk_loader/label.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,50 +63,55 @@ def update_node_dictionary(self, identifier):

def process_entities(self):
entities_created = 0
with click.progressbar(
self.reader,
length=self.entities_count,
label=self.entity_str,
update_min_steps=100,
) as reader:
for row in reader:
self.validate_row(row)
try:
with click.progressbar(
self.reader,
length=self.entities_count,
label=self.entity_str,
update_min_steps=100,
) as reader:
Comment on lines +66 to +72

Copilot AI Apr 29, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

try/finally here guarantees closure for exceptions during row processing, but it still won’t close the file if an exception is raised during EntityFile.__init__ (e.g., invalid/empty header in convert_header() or errors in count_entities()), since the handle is opened in the base constructor and parse_schemas() doesn’t guard construction. Consider moving file opening into process_entities() (using a context manager) or wrapping the base-class init work in a try/except that closes self.infile before re-raising.

Copilot uses AI. Check for mistakes.
for row in reader:
self.validate_row(row)

# Update the node identifier dictionary if necessary
if self.config.store_node_identifiers:
id_field = row[self.id]
if self.id_namespace is not None:
id_field = self.id_namespace + "." + str(id_field)
self.update_node_dictionary(id_field)
# Update the node identifier dictionary if necessary
if self.config.store_node_identifiers:
id_field = row[self.id]
if self.id_namespace is not None:
id_field = self.id_namespace + "." + str(id_field)
self.update_node_dictionary(id_field)

try:
row_binary = self.pack_props(row)
except SchemaError as e:
# TODO why is line_num off by one?
raise SchemaError(
"%s:%d %s"
% (self.infile.name, self.reader.line_num - 1, str(e))
)
row_binary_len = len(row_binary)
# If the addition of this entity will make the binary token grow too large,
# send the buffer now.
# TODO how much of this can be made uniform w/ relations and moved to Querybuffer?
added_size = self.binary_size + row_binary_len
if (
added_size >= self.config.max_token_size
or self.query_buffer.buffer_size + added_size
>= self.config.max_buffer_size
):
self.query_buffer.labels.append(self.to_binary())
self.query_buffer.send_buffer()
self.reset_partial_binary()
# Push the label onto the query buffer again, as there are more entities to process.
self.query_buffer.labels.append(self.to_binary())
try:
row_binary = self.pack_props(row)
except SchemaError as e:
# TODO why is line_num off by one?
raise SchemaError(
"%s:%d %s"
% (self.infile.name, self.reader.line_num - 1, str(e))
) from e
row_binary_len = len(row_binary)
# If the addition of this entity will make the binary token grow too large,
# send the buffer now.
# TODO how much of this can be made uniform w/ relations and moved to Querybuffer?
added_size = self.binary_size + row_binary_len
if (
added_size >= self.config.max_token_size
or self.query_buffer.buffer_size + added_size
>= self.config.max_buffer_size
):
self.query_buffer.labels.append(self.to_binary())
self.query_buffer.send_buffer()
self.reset_partial_binary()
# Push the label onto the query buffer again, as there are more entities to process.
self.query_buffer.labels.append(self.to_binary())

self.query_buffer.node_count += 1
entities_created += 1
self.binary_size += row_binary_len
self.binary_entities.append(row_binary)
self.query_buffer.labels.append(self.to_binary())
self.infile.close()
self.query_buffer.node_count += 1
entities_created += 1
self.binary_size += row_binary_len
self.binary_entities.append(row_binary)
self.query_buffer.labels.append(self.to_binary())
finally:
try:
self.infile.close()
except OSError:
Comment thread
github-code-quality[bot] marked this conversation as resolved.
Fixed
pass
print("%d nodes created with label '%s'" % (entities_created, self.entity_str))
118 changes: 62 additions & 56 deletions falkordb_bulk_loader/relation_type.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,65 +55,71 @@ def post_process_header_with_schema(self, header):

def process_entities(self):
entities_created = 0
with click.progressbar(
self.reader,
length=self.entities_count,
label=self.entity_str,
update_min_steps=100,
) as reader:
for row in reader:
self.validate_row(row)
try:
start_id = row[self.start_id]
if self.start_namespace:
start_id = self.start_namespace + "." + str(start_id)
end_id = row[self.end_id]
if self.end_namespace:
end_id = self.end_namespace + "." + str(end_id)
try:
with click.progressbar(
self.reader,
length=self.entities_count,
label=self.entity_str,
update_min_steps=100,
) as reader:
for row in reader:
self.validate_row(row)
try:
start_id = row[self.start_id]
if self.start_namespace:
start_id = self.start_namespace + "." + str(start_id)
end_id = row[self.end_id]
if self.end_namespace:
end_id = self.end_namespace + "." + str(end_id)

src = self.query_buffer.nodes[start_id]
dest = self.query_buffer.nodes[end_id]
except KeyError:
print(
"%s:%d Relationship specified a non-existent identifier. src: %s; dest: %s"
% (
self.infile.name,
self.reader.line_num - 1,
row[self.start_id],
row[self.end_id],
src = self.query_buffer.nodes[start_id]
dest = self.query_buffer.nodes[end_id]
except KeyError:
print(
"%s:%d Relationship specified a non-existent identifier. src: %s; dest: %s"
% (
self.infile.name,
self.reader.line_num - 1,
row[self.start_id],
row[self.end_id],
)
)
)
if self.config.skip_invalid_edges is False:
raise
continue
fmt = "=QQ" # 8-byte unsigned ints for src and dest
try:
row_binary = struct.pack(fmt, src, dest) + self.pack_props(row)
except SchemaError as e:
raise SchemaError(
"%s:%d %s" % (self.infile.name, self.reader.line_num, str(e))
)
row_binary_len = len(row_binary)
# If the addition of this entity will make the binary token grow too large,
# send the buffer now.
added_size = self.binary_size + row_binary_len
if (
added_size >= self.config.max_token_size
or self.query_buffer.buffer_size + added_size
>= self.config.max_buffer_size
):
self.query_buffer.reltypes.append(self.to_binary())
self.query_buffer.send_buffer()
self.reset_partial_binary()
# Push the reltype onto the query buffer again, as there are more entities to process.
self.query_buffer.reltypes.append(self.to_binary())
if self.config.skip_invalid_edges is False:
raise
continue
fmt = "=QQ" # 8-byte unsigned ints for src and dest
try:
row_binary = struct.pack(fmt, src, dest) + self.pack_props(row)
except SchemaError as e:
raise SchemaError(
"%s:%d %s"
% (self.infile.name, self.reader.line_num, str(e))
) from e
row_binary_len = len(row_binary)
# If the addition of this entity will make the binary token grow too large,
# send the buffer now.
added_size = self.binary_size + row_binary_len
if (
added_size >= self.config.max_token_size
or self.query_buffer.buffer_size + added_size
>= self.config.max_buffer_size
):
self.query_buffer.reltypes.append(self.to_binary())
self.query_buffer.send_buffer()
self.reset_partial_binary()
# Push the reltype onto the query buffer again, as there are more entities to process.
self.query_buffer.reltypes.append(self.to_binary())

self.query_buffer.relation_count += 1
entities_created += 1
self.binary_size += row_binary_len
self.binary_entities.append(row_binary)
self.query_buffer.reltypes.append(self.to_binary())
self.infile.close()
self.query_buffer.relation_count += 1
entities_created += 1
self.binary_size += row_binary_len
self.binary_entities.append(row_binary)
self.query_buffer.reltypes.append(self.to_binary())
finally:
try:
self.infile.close()
except OSError:
Comment thread
github-code-quality[bot] marked this conversation as resolved.
Fixed
pass
print(
"%d relations created for type '%s'" % (entities_created, self.entity_str)
)
Loading