def ingest_report()

in backend/code_coverage_backend/gcp.py [0:0]
33 lines of code
7 McCabe index (conditional complexity)

    def ingest_report(self, report: Report) -> bool:
        """
        When a report exist for a changeset, download it and update redis data
        """
        # Download the report
        if not download_report(self.reports_dir, self.bucket, report.name):
            logger.info("Report not available", report=str(report))
            return False

        # Read overall coverage for history
        data = covdir.open_report(report.path)
        assert data is not None, "No report to ingest"
        overall_coverage = covdir.get_overall_coverage(data)
        assert len(overall_coverage) > 0, "No overall coverage"
        self.redis.hmset(report.key_overall, overall_coverage)

        # Apply expiry for overall report
        if report.ttl is not None:
            self.redis.expire(report.key_overall, report.ttl)

        # Add the changeset to the sorted sets of known reports
        # The numeric push_id is used as a score to keep the ingested
        # changesets ordered
        self.redis.zadd(
            KEY_REPORTS.format(
                repository=report.repository,
                platform=report.platform,
                suite=report.suite,
            ),
            {report.changeset: report.push_id},
        )

        # Add the changeset to the sorted sets of known reports by date
        self.redis.zadd(
            KEY_HISTORY.format(repository=report.repository),
            {report.changeset: report.date},
        )

        # Store the filters
        if report.platform != DEFAULT_FILTER:
            self.redis.sadd(
                KEY_PLATFORMS.format(repository=report.repository), report.platform
            )
        if report.suite != DEFAULT_FILTER:
            self.redis.sadd(
                KEY_SUITES.format(repository=report.repository), report.suite
            )

        logger.info("Ingested report", report=str(report))
        return True