Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
return v.value
if isinstance(col.type, JSONEncodedDict):
return json.dumps(v)
if isinstance(v, (datetime, date)):
return v.isoformat()
return v
keys = [col.name for col in table.columns]
cols = {col.name: col for col in table.columns}
rows = [keys]
for row in DBSession.execute(select([table])):
rows.append([conv(row[key], cols[key]) for key in keys])
if len(rows) > 1:
with UnicodeWriter(fpath) as writer:
writer.writerows(rows)
{
"columnReference": "DomainElement",
"reference": {
"resource": fname.name,
"columnReference": "ID"
}
}
]
header = ['ID', 'Name', 'Numeric', 'Description']
ds.metadata['tables'].append({
'url': fname.name,
'tableSchema': {
'columns': [{'name': n, 'datatype': 'string'} for n in header]
}
})
with UnicodeWriter(fname) as writer:
writer.writerow(header)
for de in sorted(domain, key=lambda d: (d.parameter_pk, d.number)):
writer.writerow([de.id, de.name, de.number, de.description])
ds.write(outdir)