(file_handle)
Generator for reading a JSONL (json lines) file.

    Args:
        file_handle: a file-like object containing the CSV content.

    Raises:
        RuntimeError: if there are missing fields.
        DataIngestionError: If the ingestion fails.

    Yields:
        A dict that's ready to add to the datastore.
Generator for reading a JSONL (json lines) file.
Generator for reading a JSONL ( json lines ) file .
def read_and_validate_jsonl(file_handle):
    """Generator for reading a JSONL (json lines) file.

    Args:
        file_handle: a file-like object containing the CSV content.

    Raises:
        RuntimeError: if there are missing fields.
        DataIngestionError: If the ingestion fails.

    Yields:
        A dict that's ready to add to the datastore.
    """
    # Fields that must be present in each entry of the JSONL file.
    mandatory_fields = ['message', 'datetime', 'timestamp_desc']
    lineno = 0
    for line in file_handle:
        lineno += 1
        try:
            linedict = json.loads(line)
            ld_keys = linedict.keys()
            if 'datetime' not in ld_keys and 'timestamp' in ld_keys:
                epoch = int(str(linedict['timestamp'])[:10])
                dt = datetime.datetime.fromtimestamp(epoch)
                linedict['datetime'] = dt.isoformat()
            if 'timestamp' not in ld_keys and 'datetime' in ld_keys:
                try:
                    linedict['timestamp'] = int(parser.parse(
                        linedict['datetime']).timestamp() * 1000000)
                except parser.ParserError:
                    logger.error(
                        'Unable to parse timestamp, skipping line '
                        '{0:d}'.format(lineno), exc_info=True)
                    continue

            missing_fields = [x for x in mandatory_fields if x not in linedict]
            if missing_fields:
                raise RuntimeError(
                    'Missing field(s) at line {0:n}: {1:s}'.format(
                        lineno, ','.join(missing_fields)))

            if 'tag' in linedict:
                linedict['tag'] = [
                    x for x in _parse_tag_field(linedict['tag']) if x]
            _scrub_special_tags(linedict)
            yield linedict

        except ValueError as e:
            raise errors.DataIngestionError(
                'Error parsing JSON at line {0:n}: {1:s}'.format(
                    lineno, str(e)))
def read_and_validate_jsonl ( file_handle ) : # Fields that must be present in each entry of the JSONL file. mandatory_fields = [ 'message' , 'datetime' , 'timestamp_desc' ] lineno = 0 for line in file_handle : lineno += 1 try : linedict = json . loads ( line ) ld_keys = linedict . keys ( ) if 'datetime' not in ld_keys and 'timestamp' in ld_keys : epoch = int ( str ( linedict [ 'timestamp' ] ) [ : 10 ] ) dt = datetime . datetime . fromtimestamp ( epoch ) linedict [ 'datetime' ] = dt . isoformat ( ) if 'timestamp' not in ld_keys and 'datetime' in ld_keys : try : linedict [ 'timestamp' ] = int ( parser . parse ( linedict [ 'datetime' ] ) . timestamp ( ) * 1000000 ) except parser . ParserError : logger . error ( 'Unable to parse timestamp, skipping line ' '{0:d}' . format ( lineno ) , exc_info = True ) continue missing_fields = [ x for x in mandatory_fields if x not in linedict ] if missing_fields : raise RuntimeError ( 'Missing field(s) at line {0:n}: {1:s}' . format ( lineno , ',' . join ( missing_fields ) ) ) if 'tag' in linedict : linedict [ 'tag' ] = [ x for x in _parse_tag_field ( linedict [ 'tag' ] ) if x ] _scrub_special_tags ( linedict ) yield linedict except ValueError as e : raise errors . DataIngestionError ( 'Error parsing JSON at line {0:n}: {1:s}' . format ( lineno , str ( e ) ) )