"""Returns cluster activity historical metrics."""
        start_date = _safe_parse_datetime(request.args.get("start_date"))
        end_date = _safe_parse_datetime(request.args.get("end_date"))
        with create_session() as session:
            
            dag_run_types = session.execute(
                select(DagRun.run_type, func.count(DagRun.run_id))
                .where(
                    DagRun.start_date >= start_date,
                    func.coalesce(DagRun.end_date, datetime.datetime.utcnow()) <= end_date,
                )
                .group_by(DagRun.run_type)
            ).all()
            dag_run_states = session.execute(
                select(DagRun.state, func.count(DagRun.run_id))
                .where(
                    DagRun.start_date >= start_date,
                    func.coalesce(DagRun.end_date, datetime.datetime.utcnow()) <= end_date,
                )
                .group_by(DagRun.state)
            ).all()
            
            task_instance_states = session.execute(