manage.py 文件源码

def load_dump(collection_dump, collection_name):
    """
    Restore an index from a dump file.

    :param collection_dump: Path to a local gzipped dump to load.
    :param collection_name: Name for the local index to restore the dump to. Optional; will be derived from the dump name, at your own risk. Note that the pipeline will add a "owa_" prefix string to the collection name, to ensure the proper mapping and settings are applied.
    """
    available_dumps = glob(os.path.join(LOCAL_DUMPS_DIR, '*/*.gz'))
    if not collection_dump:
        choices = []
        for i, dump in enumerate(available_dumps):
            choices.append(unicode(i+1))
            click.secho('{i}) {dump}'.format(i=i+1, dump=dump), fg='green')
        dump_idx = click.prompt('Choose one of the dumps listed above',
                                type=click.Choice(choices))
        collection_dump = available_dumps[int(dump_idx) - 1]

    collection = os.path.abspath(collection_dump)
    collection_id = '_'.join(collection.split('/')[-1].split('.')[0].split('_')[:2])

    if not collection_name:
        collection_name = collection_id.replace('owa_', '')

    source_definition = {
        'id': collection_id,
        'extractor': 'ocd_backend.extractors.staticfile.StaticJSONDumpExtractor',
        'transformer': 'ocd_backend.transformers.BaseTransformer',
        'loader': 'ocd_backend.loaders.ElasticsearchLoader',
        'item': 'ocd_backend.items.LocalDumpItem',
        'dump_path': collection,
        'index_name': collection_name
    }

    click.secho(str(source_definition), fg='yellow')

    setup_pipeline(source_definition)

    click.secho('Queued items from {}. Please make sure your Celery workers'
                ' are running, so the loaded items are processed.'.format(collection),
                fg='green')


# Register commands explicitly with groups, so we can easily use the docstring
# wrapper