Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def row_counter(self, resource, iterator):
counter = 0
for row in iterator:
counter += 1
yield row
DumperBase.inc_attr(self.datapackage.descriptor, self.datapackage_rowcount, counter)
DumperBase.inc_attr(resource.res.descriptor, self.resource_rowcount, counter)
resource.res.commit()
self.datapackage.commit()
if self.temporal_format_property:
for resource in self.datapackage.descriptor['resources']:
for field in resource['schema']['fields']:
if field.get('type') in ['datetime', 'date', 'time']:
format = field.pop(self.temporal_format_property, None)
if format:
field['format'] = format
self.datapackage.commit()
temp_file = tempfile.NamedTemporaryFile(mode="w+", delete=False, encoding='utf-8')
indent = 2 if self.pretty_descriptor else None
json.dump(self.datapackage.descriptor, temp_file, indent=indent, sort_keys=True, ensure_ascii=False)
temp_file_name = temp_file.name
filesize = temp_file.tell()
temp_file.close()
DumperBase.inc_attr(self.datapackage.descriptor, self.datapackage_bytes, filesize)
self.write_file_to_output(temp_file_name, 'datapackage.json')
# if location is not None:
# stats.setdefault(STATS_DPP_KEY, {})[STATS_OUT_DP_URL_KEY] = location
os.unlink(temp_file_name)
super(FileDumper, self).handle_datapackage()
def rows_processor(self, resource, writer, temp_file):
for row in resource:
writer.write_row(row)
yield row
writer.finalize_file()
# Get resource descriptor
resource_descriptor = resource.res.descriptor
for descriptor in self.datapackage.descriptor['resources']:
if descriptor['name'] == resource.res.descriptor['name']:
resource_descriptor = descriptor
# File size:
filesize = temp_file.tell()
DumperBase.inc_attr(self.datapackage.descriptor, self.datapackage_bytes, filesize)
DumperBase.inc_attr(resource_descriptor, self.resource_bytes, filesize)
# File Hash:
if self.resource_hash:
hasher = FileDumper.hash_handler(temp_file)
# Update path with hash
if self.add_filehash_to_path:
DumperBase.insert_hash_in_path(resource_descriptor, hasher.hexdigest())
DumperBase.set_attr(resource_descriptor, self.resource_hash, hasher.hexdigest())
# Finalise
filename = temp_file.name
temp_file.close()
self.write_file_to_output(filename, resource.res.source)
os.unlink(filename)
def rows_processor(self, resource, writer, temp_file):
for row in resource:
writer.write_row(row)
yield row
writer.finalize_file()
# Get resource descriptor
resource_descriptor = resource.res.descriptor
for descriptor in self.datapackage.descriptor['resources']:
if descriptor['name'] == resource.res.descriptor['name']:
resource_descriptor = descriptor
# File size:
filesize = temp_file.tell()
DumperBase.inc_attr(self.datapackage.descriptor, self.datapackage_bytes, filesize)
DumperBase.inc_attr(resource_descriptor, self.resource_bytes, filesize)
# File Hash:
if self.resource_hash:
hasher = FileDumper.hash_handler(temp_file)
# Update path with hash
if self.add_filehash_to_path:
DumperBase.insert_hash_in_path(resource_descriptor, hasher.hexdigest())
DumperBase.set_attr(resource_descriptor, self.resource_hash, hasher.hexdigest())
# Finalise
filename = temp_file.name
temp_file.close()
self.write_file_to_output(filename, resource.res.source)
os.unlink(filename)