Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def generate_data(self, session, engine, base, num_records, current_batch_num):
output_stream = SqlOutputStream.from_connection(
session, engine, base, self.mappings
)
with open(self.yaml_file) as open_yaml_file:
summary = generate(
open_yaml_file, self.num_records, self.vars, output_stream
)
output_stream.close()
if self.generate_mapping_file:
with open(self.generate_mapping_file, "w+") as f:
yaml.safe_dump(
mapping_from_factory_templates(summary), f, sort_keys=False
)
f.close()