(table)
| 1170 | @app.route('/<table>/import/', methods=['GET', 'POST']) |
| 1171 | @require_table |
| 1172 | def table_import(table): |
| 1173 | dataset = get_dataset() |
| 1174 | count = None |
| 1175 | request_data = get_request_data() |
| 1176 | strict = bool(request_data.get('strict')) |
| 1177 | |
| 1178 | if request.method == 'POST': |
| 1179 | file_obj = request.files.get('file') |
| 1180 | if not file_obj: |
| 1181 | flash('Please select an import file.', 'danger') |
| 1182 | elif not file_obj.filename.lower().endswith(('.csv', '.json')): |
| 1183 | flash('Unsupported file-type. Must be a .json or .csv file.', |
| 1184 | 'danger') |
| 1185 | else: |
| 1186 | if file_obj.filename.lower().endswith('.json'): |
| 1187 | format = 'json' |
| 1188 | else: |
| 1189 | format = 'csv' |
| 1190 | |
| 1191 | # Here we need to translate the file stream. Werkzeug uses a |
| 1192 | # spooled temporary file opened in wb+ mode, which is not |
| 1193 | # compatible with Python's CSV module. We'd need to reach pretty |
| 1194 | # far into Flask's internals to modify this behavior, so instead |
| 1195 | # we'll just translate the stream into utf8-decoded unicode. |
| 1196 | try: |
| 1197 | stream = TextIOWrapper(file_obj, encoding='utf8') |
| 1198 | except AttributeError: |
| 1199 | # The SpooledTemporaryFile used by werkzeug does not |
| 1200 | # implement an API that the TextIOWrapper expects, so we'll |
| 1201 | # just consume the whole damn thing and decode it. |
| 1202 | # Fixed in werkzeug 0.15. |
| 1203 | stream = StringIO(file_obj.read().decode('utf8')) |
| 1204 | |
| 1205 | kwargs = {} |
| 1206 | if peewee_version >= (4, 0, 2): |
| 1207 | kwargs['base64_bytes'] = app.config['BLOB_AS_BASE64'] |
| 1208 | try: |
| 1209 | with dataset.transaction(): |
| 1210 | count = dataset.thaw( |
| 1211 | table, |
| 1212 | format=format, |
| 1213 | file_obj=stream, |
| 1214 | strict=strict, |
| 1215 | **kwargs) |
| 1216 | except Exception as exc: |
| 1217 | flash('Error importing file: %s' % exc, 'danger') |
| 1218 | app.logger.exception('Error importing file.') |
| 1219 | else: |
| 1220 | flash( |
| 1221 | 'Successfully imported %s objects from %s.' % ( |
| 1222 | count, file_obj.filename), |
| 1223 | 'success') |
| 1224 | return redirect(url_for('table_content', table=table)) |
| 1225 | |
| 1226 | return render_template( |
| 1227 | 'table_import.html', |
| 1228 | count=count, |
| 1229 | strict=strict, |
nothing calls this directly
no test coverage detected