def verify_doc_count()

in src/es_pii_tool/base.py [0:0]

• 46 lines of code
• 4 McCabe index (conditional complexity)

    def verify_doc_count(self, job: Job) -> bool:
        """Verify that expected_docs and the hits from the query have the same value

        :param job: The job object for the present redaction run

        :type job: :py:class:`~.app.tracking.Job`

        :rtype: None
        :returns: No return value
        """
        try:
            task = Task(job, task_id=f'PRE---{job.name}---DOC-COUNT-VERIFICATION')
        except Exception as err:
            logger.critical('Unable to create task: %s', err)
            raise FatalError('Unable to create task', err) from err
        success = False
        errors = False
        if task.finished():
            return True  # We're done already
        # Log task start
        task.begin()
        hits = 0
        try:
            hits = get_hits(self.client, job.config['pattern'], job.config['query'])
        except Exception as err:
            logger.critical('Unable to count query result hits: %s', err)
            raise err
        msg = f'{hits} hit(s)'
        logger.debug(msg)
        task.add_log(msg)
        logger.info("Checking expected document count...")
        zeromsg = (
            f"For index pattern {job.config['pattern']}, with query "
            f"{job.config['query']} 'expected_docs' is {job.config['expected_docs']} "
            f"but query results is {hits} matches."
        )
        if job.config['expected_docs'] == hits:
            msg = (
                f'Query result hits: {hits} matches expected_docs: '
                f'{job.config["expected_docs"]}'
            )
            logger.debug(msg)
            task.add_log(msg)
            success = True
            if hits == 0:
                logger.critical(zeromsg)
                logger.info('Continuing to next configuration block (if any)')
                success = False
        else:
            logger.critical(zeromsg)
            logger.info('Continuing to next configuration block (if any)')
        if not success:
            errors = True
            task.add_log(zeromsg)
        task.end(success, errors=errors)
        return success