(
        my_sketch: sketch.Sketch, config_dict: Dict[str, any],
        file_path: str)
return timeline, task_id
Uploads a file to Timesketch.

    Args:
        my_sketch (sketch.Sketch): a sketch object to point to the sketch the
            data will be imported to.
        config_dict (dict): dict with settings for the importer.
        file_path (str): the path to the file to upload.

    Returns:
        A tuple with the timeline object (timeline.Timeline) or None if not
        able to upload the timeline as well as the celery task identification
        for the indexing.
Uploads a file to Timesketch.
Uploads a file to Timesketch .
def upload_file(
        my_sketch: sketch.Sketch, config_dict: Dict[str, any],
        file_path: str) -> str:
    """Uploads a file to Timesketch.

    Args:
        my_sketch (sketch.Sketch): a sketch object to point to the sketch the
            data will be imported to.
        config_dict (dict): dict with settings for the importer.
        file_path (str): the path to the file to upload.

    Returns:
        A tuple with the timeline object (timeline.Timeline) or None if not
        able to upload the timeline as well as the celery task identification
        for the indexing.
    """
    if not my_sketch or not hasattr(my_sketch, 'id'):
        return 'Sketch needs to be set'

    _, _, file_extension = file_path.rpartition('.')
    if file_extension.lower() not in ('plaso', 'csv', 'jsonl'):
        return (
            'File needs to have one of the following extensions: '
            '.plaso, .csv, '
            '.jsonl (not {0:s})').format(file_extension.lower())

    import_helper = helper.ImportHelper()
    import_helper.add_config_dict(config_dict)

    log_config_file = config_dict.get('log_config_file', '')
    if log_config_file:
        import_helper.add_config(log_config_file)

    timeline = None
    task_id = ''
    logger.info('About to upload file.')
    with importer.ImportStreamer() as streamer:
        streamer.set_sketch(my_sketch)
        streamer.set_config_helper(import_helper)
        streamer.set_provider('CLI importer tool')

        format_string = config_dict.get('message_format_string')
        if format_string:
            streamer.set_message_format_string(format_string)

        timeline_name = config_dict.get('timeline_name')
        if timeline_name:
            streamer.set_timeline_name(timeline_name)

        index_name = config_dict.get('index_name')
        if index_name:
            streamer.set_index_name(index_name)

        time_desc = config_dict.get('timestamp_description')
        if time_desc:
            streamer.set_timestamp_description(time_desc)

        entry_threshold = config_dict.get('entry_threshold')
        if entry_threshold:
            streamer.set_entry_threshold(entry_threshold)

        size_threshold = config_dict.get('size_threshold')
        if size_threshold:
            streamer.set_filesize_threshold(size_threshold)

        data_label = config_dict.get('data_label')
        if data_label:
            streamer.set_data_label(data_label)

        context = config_dict.get('context')
        if context:
            streamer.set_upload_context(context)
        else:
            streamer.set_upload_context(' '.join(sys.argv))

        streamer.add_file(file_path)

        # Force a flush.
        streamer.flush()

        timeline = streamer.timeline
        task_id = streamer.celery_task_id

    logger.info('File upload completed.')
    return timeline, task_id
def upload_file ( my_sketch : sketch . Sketch , config_dict : Dict [ str , any ] , file_path : str ) -> str : if not my_sketch or not hasattr ( my_sketch , 'id' ) : return 'Sketch needs to be set' _ , _ , file_extension = file_path . rpartition ( '.' ) if file_extension . lower ( ) not in ( 'plaso' , 'csv' , 'jsonl' ) : return ( 'File needs to have one of the following extensions: ' '.plaso, .csv, ' '.jsonl (not {0:s})' ) . format ( file_extension . lower ( ) ) import_helper = helper . ImportHelper ( ) import_helper . add_config_dict ( config_dict ) log_config_file = config_dict . get ( 'log_config_file' , '' ) if log_config_file : import_helper . add_config ( log_config_file ) timeline = None task_id = '' logger . info ( 'About to upload file.' ) with importer . ImportStreamer ( ) as streamer : streamer . set_sketch ( my_sketch ) streamer . set_config_helper ( import_helper ) streamer . set_provider ( 'CLI importer tool' ) format_string = config_dict . get ( 'message_format_string' ) if format_string : streamer . set_message_format_string ( format_string ) timeline_name = config_dict . get ( 'timeline_name' ) if timeline_name : streamer . set_timeline_name ( timeline_name ) index_name = config_dict . get ( 'index_name' ) if index_name : streamer . set_index_name ( index_name ) time_desc = config_dict . get ( 'timestamp_description' ) if time_desc : streamer . set_timestamp_description ( time_desc ) entry_threshold = config_dict . get ( 'entry_threshold' ) if entry_threshold : streamer . set_entry_threshold ( entry_threshold ) size_threshold = config_dict . get ( 'size_threshold' ) if size_threshold : streamer . set_filesize_threshold ( size_threshold ) data_label = config_dict . get ( 'data_label' ) if data_label : streamer . set_data_label ( data_label ) context = config_dict . get ( 'context' ) if context : streamer . set_upload_context ( context ) else : streamer . set_upload_context ( ' ' . join ( sys . argv ) ) streamer . add_file ( file_path ) # Force a flush. streamer . flush ( ) timeline = streamer . timeline task_id = streamer . celery_task_id logger . info ( 'File upload completed.' ) return timeline , task_id