Export datasources to ZIP file
(datasource_file: Optional[str] = None)
| 303 | "--datasource-file", "-f", help="Specify the the file to export to", |
| 304 | ) |
| 305 | def export_datasources(datasource_file: Optional[str] = None) -> None: |
| 306 | """Export datasources to ZIP file""" |
| 307 | # pylint: disable=import-outside-toplevel |
| 308 | from superset.connectors.sqla.models import SqlaTable |
| 309 | from superset.datasets.commands.export import ExportDatasetsCommand |
| 310 | |
| 311 | g.user = security_manager.find_user(username="admin") |
| 312 | |
| 313 | dataset_ids = [id_ for (id_,) in db.session.query(SqlaTable.id).all()] |
| 314 | timestamp = datetime.now().strftime("%Y%m%dT%H%M%S") |
| 315 | root = f"dataset_export_{timestamp}" |
| 316 | datasource_file = datasource_file or f"{root}.zip" |
| 317 | |
| 318 | try: |
| 319 | with ZipFile(datasource_file, "w") as bundle: |
| 320 | for file_name, file_content in ExportDatasetsCommand(dataset_ids).run(): |
| 321 | with bundle.open(f"{root}/{file_name}", "w") as fp: |
| 322 | fp.write(file_content.encode()) |
| 323 | except Exception: # pylint: disable=broad-except |
| 324 | logger.exception( |
| 325 | "There was an error when exporting the datasets, please check " |
| 326 | "the exception traceback in the log" |
| 327 | ) |
| 328 | |
| 329 | @superset.command() |
| 330 | @with_appcontext |
nothing calls this directly
no test coverage detected