redash_toolbelt/examples/migrate.py
def import_data_sources(orig_client, dest_client):
    """Create stub data sources at DESTINATION based on the data sources in ORIGIN."""

    orig_data_sources = orig_client.get_data_sources()
    allowed_types = {
        i["type"]: i for i in dest_client._get("api/data_sources/types").json()
    }

    migratable_types = {"csvurl": {"name": "csv", "command-name": "fix-csv-queries"}}

    for ds in orig_data_sources:

        dsid = ds["id"]

        if dsid in meta["data_sources"]:
            print(
                "WARNING: origin data source {} -- SKIP -- already exists".format(dsid)
            )
            continue

        ds_def = orig_client.get_data_source(dsid)

        origin_type = ds_def["type"]

        is_allowed_type = origin_type in allowed_types
        is_migratable_type = origin_type in migratable_types

        if not is_allowed_type and not is_migratable_type:
            print(
                "WARNING: origin data source {} -- SKIP -- destination instance does not support this type: {}".format(
                    dsid, ds_def["type"]
                )
            )
            continue

        if is_allowed_type:
            target_type = origin_type
        elif is_migratable_type:
            target_type = migratable_types[origin_type]["name"]
            print(
                textwrap.dedent(
                    """
                INFO: Origin data source type {} is not supported, but queries can be mapped to type {}.
                Run {} to update query text for this new query runner.""".format(
                        origin_type,
                        target_type,
                        migratable_types[origin_type]["command-name"],
                    )
                )
            )

        new_data_source_def = {
            "name": ds_def["name"],
            "_type": target_type,
            "options": ds_def["options"],
        }

        try:
            new_ds = dest_client.create_data_source(**new_data_source_def).json()
            meta["data_sources"][dsid] = new_ds["id"]
        except Exception as e:
            print(
                "ERROR: origin data source {} -- FAIL -- could not be imported: {}".format(
                    dsid, str(e)
                )
            )
            continue
Create stub data sources at DESTINATION based on the data sources in ORIGIN.
Create stub data sources at DESTINATION based on the data sources in ORIGIN.
Create stub data sources at DESTINATION based on the data sources in ORIGIN .
def import_data_sources(orig_client, dest_client):
    orig_data_sources = orig_client.get_data_sources()
    allowed_types = {
        i["type"]: i for i in dest_client._get("api/data_sources/types").json()
    }
    migratable_types = {"csvurl": {"name": "csv", "command-name": "fix-csv-queries"}}
    for ds in orig_data_sources:
        dsid = ds["id"]
        if dsid in meta["data_sources"]:
            print(
                "WARNING: origin data source {} -- SKIP -- already exists".format(dsid)
            )
            continue
        ds_def = orig_client.get_data_source(dsid)
        origin_type = ds_def["type"]
        is_allowed_type = origin_type in allowed_types
        is_migratable_type = origin_type in migratable_types
        if not is_allowed_type and not is_migratable_type:
            print(
                "WARNING: origin data source {} -- SKIP -- destination instance does not support this type: {}".format(
                    dsid, ds_def["type"]
                )
            )
            continue
        if is_allowed_type:
            target_type = origin_type
        elif is_migratable_type:
            target_type = migratable_types[origin_type]["name"]
            print(
                textwrap.dedent(
                    """
                INFO: Origin data source type {} is not supported, but queries can be mapped to type {}.
                Run {} to update query text for this new query runner.""".format(
                        origin_type,
                        target_type,
                        migratable_types[origin_type]["command-name"],
                    )
                )
            )
        new_data_source_def = {
            "name": ds_def["name"],
            "_type": target_type,
            "options": ds_def["options"],
        }
        try:
            new_ds = dest_client.create_data_source(**new_data_source_def).json()
            meta["data_sources"][dsid] = new_ds["id"]
        except Exception as e:
            print(
                "ERROR: origin data source {} -- FAIL -- could not be imported: {}".format(
                    dsid, str(e)
                )
            )
            continue
def import_data_sources ( orig_client , dest_client ) : orig_data_sources = orig_client . get_data_sources ( ) allowed_types = { i [ "type" ] : i for i in dest_client . _get ( "api/data_sources/types" ) . json ( ) } migratable_types = { "csvurl" : { "name" : "csv" , "command-name" : "fix-csv-queries" } } for ds in orig_data_sources : dsid = ds [ "id" ] if dsid in meta [ "data_sources" ] : print ( "WARNING: origin data source {} -- SKIP -- already exists" . format ( dsid ) ) continue ds_def = orig_client . get_data_source ( dsid ) origin_type = ds_def [ "type" ] is_allowed_type = origin_type in allowed_types is_migratable_type = origin_type in migratable_types if not is_allowed_type and not is_migratable_type : print ( "WARNING: origin data source {} -- SKIP -- destination instance does not support this type: {}" . format ( dsid , ds_def [ "type" ] ) ) continue if is_allowed_type : target_type = origin_type elif is_migratable_type : target_type = migratable_types [ origin_type ] [ "name" ] print ( textwrap . dedent ( """
                INFO: Origin data source type {} is not supported, but queries can be mapped to type {}.
                Run {} to update query text for this new query runner.""" . format ( origin_type , target_type , migratable_types [ origin_type ] [ "command-name" ] , ) ) ) new_data_source_def = { "name" : ds_def [ "name" ] , "_type" : target_type , "options" : ds_def [ "options" ] , } try : new_ds = dest_client . create_data_source ( ** new_data_source_def ) . json ( ) meta [ "data_sources" ] [ dsid ] = new_ds [ "id" ] except Exception as e : print ( "ERROR: origin data source {} -- FAIL -- could not be imported: {}" . format ( dsid , str ( e ) ) ) continue
Create stub data sources at DESTINATION based on the data sources in ORIGIN.
Create stub data sources at DESTINATION based on the data sources in ORIGIN .
"""Create stub data sources at DESTINATION based on the data sources in ORIGIN."""
orig_client
dest_client