def get_homepage_stats()

in backend/services/stats_service.py [0:0]

• 118 lines of code
• 8 McCabe index (conditional complexity)

    def get_homepage_stats(abbrev=True) -> HomePageStatsDTO:
        """ Get overall TM stats to give community a feel for progress that's being made """
        dto = HomePageStatsDTO()
        dto.total_projects = Project.query.with_entities(
            func.count(Project.id)
        ).scalar()
        dto.mappers_online = (
            Task.query.with_entities(func.count(Task.locked_by.distinct()))
            .filter(Task.locked_by.isnot(None))
            .scalar()
        )
        dto.total_mappers = User.query.with_entities(func.count(User.id)).scalar()
        dto.tasks_mapped = (
            Task.query.with_entities(func.count())
            .filter(
                Task.task_status.in_(
                    (TaskStatus.MAPPED.value, TaskStatus.VALIDATED.value)
                )
            )
            .scalar()
        )
        if not abbrev:
            dto.total_validators = (
                Task.query.filter(Task.task_status == TaskStatus.VALIDATED.value)
                .distinct(Task.validated_by)
                .count()
            )
            dto.tasks_validated = Task.query.filter(
                Task.task_status == TaskStatus.VALIDATED.value
            ).count()

            dto.total_area = Project.query.with_entities(
                func.coalesce(func.sum(func.ST_Area(Project.geometry, True) / 1000000))
            ).scalar()

            dto.total_mapped_area = (
                Task.query.with_entities(
                    func.coalesce(func.sum(func.ST_Area(Task.geometry, True) / 1000000))
                )
                .filter(Task.task_status == TaskStatus.MAPPED.value)
                .scalar()
            )

            dto.total_validated_area = (
                Task.query.with_entities(
                    func.coalesce(func.sum(func.ST_Area(Task.geometry, True) / 1000000))
                )
                .filter(Task.task_status == TaskStatus.VALIDATED.value)
                .scalar()
            )

            unique_campaigns = Campaign.query.with_entities(
                func.count(Campaign.id)
            ).scalar()

            linked_campaigns_count = (
                Campaign.query.join(
                    campaign_projects, Campaign.id == campaign_projects.c.campaign_id
                )
                .with_entities(
                    Campaign.name, func.count(campaign_projects.c.campaign_id)
                )
                .group_by(Campaign.id)
                .all()
            )

            subquery = (
                db.session.query(campaign_projects.c.project_id.distinct())
                .order_by(campaign_projects.c.project_id)
                .subquery()
            )
            no_campaign_count = (
                Project.query.with_entities(func.count())
                .filter(~Project.id.in_(subquery))
                .scalar()
            )
            dto.campaigns = [CampaignStatsDTO(row) for row in linked_campaigns_count]
            if no_campaign_count:
                dto.campaigns.append(
                    CampaignStatsDTO(("Unassociated", no_campaign_count))
                )

            dto.total_campaigns = unique_campaigns
            unique_orgs = Organisation.query.with_entities(
                func.count(Organisation.id)
            ).scalar()

            linked_orgs_count = (
                db.session.query(Organisation.name, func.count(Project.organisation_id))
                .join(Project.organisation)
                .group_by(Organisation.id)
                .all()
            )

            subquery = (
                db.session.query(Project.organisation_id.distinct())
                .order_by(Project.organisation_id)
                .subquery()
            )
            no_org_project_count = (
                Organisation.query.with_entities(func.count())
                .filter(~Organisation.id.in_(subquery))
                .scalar()
            )
            dto.organisations = [
                OrganizationListStatsDTO(row) for row in linked_orgs_count
            ]

            if no_org_project_count:
                no_org_proj = OrganizationListStatsDTO(
                    ("Unassociated", no_org_project_count)
                )
                dto.organisations.append(no_org_proj)

            dto.total_organisations = unique_orgs
        else:
            # Clear null attributes for abbreviated call
            clear_attrs = [
                "total_validators",
                "tasks_validated",
                "total_area",
                "total_mapped_area",
                "total_validated_area",
                "campaigns",
                "total_campaigns",
                "organisations",
                "total_organisations",
            ]

            for attr in clear_attrs:
                delattr(dto, attr)

        return dto