(self, sketch_id)
return response
Handles POST request to the resource.

        Args:
            sketch_id: Integer primary key for a sketch database model

        Returns:
            An annotation in JSON (instance of flask.wrappers.Response)
Handles POST request to the resource.
Handles POST request to the resource .
def post(self, sketch_id):
        """Handles POST request to the resource.

        Args:
            sketch_id: Integer primary key for a sketch database model

        Returns:
            An annotation in JSON (instance of flask.wrappers.Response)
        """
        sketch = Sketch.query.get_with_acl(sketch_id)
        if not sketch:
            msg = 'No sketch found with this ID.'
            abort(HTTP_STATUS_CODE_NOT_FOUND, msg)

        if not sketch.has_permission(current_user, 'write'):
            abort(
                HTTP_STATUS_CODE_FORBIDDEN, (
                    'User does not have sufficient access rights to '
                    'modify a sketch.'))

        form = request.json
        if not form:
            form = request.data

        tag_dict = {
            'events_processed_by_api': 0,
            'number_of_events_with_added_tags': 0,
            'tags_applied': 0,
        }
        datastore = self.datastore

        try:
            tags_to_add = json.loads(form.get('tag_string', ''))
        except json.JSONDecodeError as e:
            abort(
                HTTP_STATUS_CODE_BAD_REQUEST,
                'Unable to read the tags, with error: {0!s}'.format(e))

        if not isinstance(tags_to_add, list):
            abort(
                HTTP_STATUS_CODE_BAD_REQUEST, 'Tags need to be a list')

        if not all(isinstance(x, str) for x in tags_to_add):
            abort(
                HTTP_STATUS_CODE_BAD_REQUEST,
                'Tags need to be a list of strings')

        events = form.get('events', [])
        event_df = pd.DataFrame(events)

        for field in ['_id', '_type', '_index']:
            if field not in event_df:
                abort(
                    HTTP_STATUS_CODE_BAD_REQUEST,
                    'Events need to have a [{0:s}] field associated '
                    'to it.'.format(field))
            if any(event_df[field].isna()):
                abort(
                    HTTP_STATUS_CODE_BAD_REQUEST,
                    'All events need to have a [{0:s}] field '
                    'set, it cannot have a non-value.'.format(field))

        # Remove any potential extra fields from the events.
        event_df = event_df[['_id', '_type', '_index']]

        tag_df = pd.DataFrame()

        event_size = event_df.shape[0]
        if event_size > self.MAX_EVENTS_TO_TAG:
            abort(
                HTTP_STATUS_CODE_BAD_REQUEST,
                'Cannot tag more than {0:d} events in a single '
                'request'.format(self.MAX_EVENTS_TO_TAG))

        tag_dict['number_of_events_passed_to_api'] = event_size

        errors = []

        verbose = form.get('verbose', False)
        if verbose:
            tag_dict['number_of_indices'] = len(event_df['_index'].unique())
            time_tag_gathering_start = time.time()

        for _index in event_df['_index'].unique():
            index_slice = event_df[event_df['_index'] == _index]
            index_size = index_slice.shape[0]

            if verbose:
                tag_dict.setdefault('index_count', {})
                tag_dict['index_count'][_index] = index_size

            if index_size <= self.EVENT_CHUNK_SIZE:
                chunks = 1
            else:
                chunks = math.ceil(index_size / self.EVENT_CHUNK_SIZE)

            tags = []
            for index_chunk in np.array_split(
                    index_slice['_id'].unique(), chunks):
                should_list = [{'match': {'_id': x}} for x in index_chunk]
                query_body = {
                    'query': {
                        'bool': {
                            'should': should_list
                        }
                    }
                }

                # Adding a small buffer to make sure all results are captured.
                size = len(should_list) + 100
                query_body['size'] = size
                query_body['terminate_after'] = size

                try:
                    # pylint: disable=unexpected-keyword-arg
                    if datastore.version.startswith('6'):
                        search = datastore.client.search(
                            body=json.dumps(query_body),
                            index=[_index],
                            _source_include=['tag'],
                            search_type='query_then_fetch'
                        )
                    else:
                        search = datastore.client.search(
                            body=json.dumps(query_body),
                            index=[_index],
                            _source_includes=['tag'],
                            search_type='query_then_fetch'
                        )

                except RequestError as e:
                    logger.error('Unable to query for events', exc_info=True)
                    errors.append(
                        'Unable to query for events, {0!s}'.format(e))
                    abort(
                        HTTP_STATUS_CODE_BAD_REQUEST,
                        'Unable to query events, {0!s}'.format(e))

                for result in search['hits']['hits']:
                    tag = result.get('_source', {}).get('tag', [])
                    if not tag:
                        continue
                    tags.append({'_id': result.get('_id'), 'tag': tag})

            if not tags:
                continue
            tag_df = pd.concat([tag_df, pd.DataFrame(tags)])

        if tag_df.shape[0]:
            event_df = event_df.merge(tag_df, on='_id', how='left')

        if verbose:
            tag_dict[
                'time_to_gather_tags'] = time.time() - time_tag_gathering_start
            tag_dict['number_of_events'] = len(events)

            if tag_df.shape[0]:
                tag_dict['number_of_events_in_tag_frame'] = tag_df.shape[0]

            if 'tag' in event_df:
                current_tag_events = event_df[~event_df['tag'].isna()].shape[0]
                tag_dict['number_of_events_with_tags'] = current_tag_events
            else:
                tag_dict['number_of_events_with_tags'] = 0

            tag_dict['tags_to_add'] = tags_to_add
            time_tag_start = time.time()

        if event_size > datastore.DEFAULT_FLUSH_INTERVAL:
            flush_interval = self.BUFFER_SIZE_FOR_ES_BULK_UPDATES
        else:
            flush_interval = datastore.DEFAULT_FLUSH_INTERVAL
        _ = event_df.apply(
            _tag_event, axis=1, tag_dict=tag_dict, tags_to_add=tags_to_add,
            datastore=datastore, flush_interval=flush_interval)
        datastore.flush_queued_events()

        if verbose:
            tag_dict['time_to_tag'] = time.time() - time_tag_start

        if errors:
            tag_dict['errors'] = errors

        schema = {
            'meta': tag_dict,
            'objects': []}
        response = jsonify(schema)
        response.status_code = HTTP_STATUS_CODE_OK
        return response
def post ( self , sketch_id ) : sketch = Sketch . query . get_with_acl ( sketch_id ) if not sketch : msg = 'No sketch found with this ID.' abort ( HTTP_STATUS_CODE_NOT_FOUND , msg ) if not sketch . has_permission ( current_user , 'write' ) : abort ( HTTP_STATUS_CODE_FORBIDDEN , ( 'User does not have sufficient access rights to ' 'modify a sketch.' ) ) form = request . json if not form : form = request . data tag_dict = { 'events_processed_by_api' : 0 , 'number_of_events_with_added_tags' : 0 , 'tags_applied' : 0 , } datastore = self . datastore try : tags_to_add = json . loads ( form . get ( 'tag_string' , '' ) ) except json . JSONDecodeError as e : abort ( HTTP_STATUS_CODE_BAD_REQUEST , 'Unable to read the tags, with error: {0!s}' . format ( e ) ) if not isinstance ( tags_to_add , list ) : abort ( HTTP_STATUS_CODE_BAD_REQUEST , 'Tags need to be a list' ) if not all ( isinstance ( x , str ) for x in tags_to_add ) : abort ( HTTP_STATUS_CODE_BAD_REQUEST , 'Tags need to be a list of strings' ) events = form . get ( 'events' , [ ] ) event_df = pd . DataFrame ( events ) for field in [ '_id' , '_type' , '_index' ] : if field not in event_df : abort ( HTTP_STATUS_CODE_BAD_REQUEST , 'Events need to have a [{0:s}] field associated ' 'to it.' . format ( field ) ) if any ( event_df [ field ] . isna ( ) ) : abort ( HTTP_STATUS_CODE_BAD_REQUEST , 'All events need to have a [{0:s}] field ' 'set, it cannot have a non-value.' . format ( field ) ) # Remove any potential extra fields from the events. event_df = event_df [ [ '_id' , '_type' , '_index' ] ] tag_df = pd . DataFrame ( ) event_size = event_df . shape [ 0 ] if event_size > self . MAX_EVENTS_TO_TAG : abort ( HTTP_STATUS_CODE_BAD_REQUEST , 'Cannot tag more than {0:d} events in a single ' 'request' . format ( self . MAX_EVENTS_TO_TAG ) ) tag_dict [ 'number_of_events_passed_to_api' ] = event_size errors = [ ] verbose = form . get ( 'verbose' , False ) if verbose : tag_dict [ 'number_of_indices' ] = len ( event_df [ '_index' ] . unique ( ) ) time_tag_gathering_start = time . time ( ) for _index in event_df [ '_index' ] . unique ( ) : index_slice = event_df [ event_df [ '_index' ] == _index ] index_size = index_slice . shape [ 0 ] if verbose : tag_dict . setdefault ( 'index_count' , { } ) tag_dict [ 'index_count' ] [ _index ] = index_size if index_size <= self . EVENT_CHUNK_SIZE : chunks = 1 else : chunks = math . ceil ( index_size / self . EVENT_CHUNK_SIZE ) tags = [ ] for index_chunk in np . array_split ( index_slice [ '_id' ] . unique ( ) , chunks ) : should_list = [ { 'match' : { '_id' : x } } for x in index_chunk ] query_body = { 'query' : { 'bool' : { 'should' : should_list } } } # Adding a small buffer to make sure all results are captured. size = len ( should_list ) + 100 query_body [ 'size' ] = size query_body [ 'terminate_after' ] = size try : # pylint: disable=unexpected-keyword-arg if datastore . version . startswith ( '6' ) : search = datastore . client . search ( body = json . dumps ( query_body ) , index = [ _index ] , _source_include = [ 'tag' ] , search_type = 'query_then_fetch' ) else : search = datastore . client . search ( body = json . dumps ( query_body ) , index = [ _index ] , _source_includes = [ 'tag' ] , search_type = 'query_then_fetch' ) except RequestError as e : logger . error ( 'Unable to query for events' , exc_info = True ) errors . append ( 'Unable to query for events, {0!s}' . format ( e ) ) abort ( HTTP_STATUS_CODE_BAD_REQUEST , 'Unable to query events, {0!s}' . format ( e ) ) for result in search [ 'hits' ] [ 'hits' ] : tag = result . get ( '_source' , { } ) . get ( 'tag' , [ ] ) if not tag : continue tags . append ( { '_id' : result . get ( '_id' ) , 'tag' : tag } ) if not tags : continue tag_df = pd . concat ( [ tag_df , pd . DataFrame ( tags ) ] ) if tag_df . shape [ 0 ] : event_df = event_df . merge ( tag_df , on = '_id' , how = 'left' ) if verbose : tag_dict [ 'time_to_gather_tags' ] = time . time ( ) - time_tag_gathering_start tag_dict [ 'number_of_events' ] = len ( events ) if tag_df . shape [ 0 ] : tag_dict [ 'number_of_events_in_tag_frame' ] = tag_df . shape [ 0 ] if 'tag' in event_df : current_tag_events = event_df [ ~ event_df [ 'tag' ] . isna ( ) ] . shape [ 0 ] tag_dict [ 'number_of_events_with_tags' ] = current_tag_events else : tag_dict [ 'number_of_events_with_tags' ] = 0 tag_dict [ 'tags_to_add' ] = tags_to_add time_tag_start = time . time ( ) if event_size > datastore . DEFAULT_FLUSH_INTERVAL : flush_interval = self . BUFFER_SIZE_FOR_ES_BULK_UPDATES else : flush_interval = datastore . DEFAULT_FLUSH_INTERVAL _ = event_df . apply ( _tag_event , axis = 1 , tag_dict = tag_dict , tags_to_add = tags_to_add , datastore = datastore , flush_interval = flush_interval ) datastore . flush_queued_events ( ) if verbose : tag_dict [ 'time_to_tag' ] = time . time ( ) - time_tag_start if errors : tag_dict [ 'errors' ] = errors schema = { 'meta' : tag_dict , 'objects' : [ ] } response = jsonify ( schema ) response . status_code = HTTP_STATUS_CODE_OK return response