def aggregate(self, table, groupby, filters={}, aggregate='count(*)', page=0, page_size=100, orderby=None):
    self.log.info((
        "table=%s, groupby=%s, filters=%s, aggregate=%s, page=%s,"
        " page_size=%s, orderby=%s"
      ), table, groupby, filters, aggregate, page, page_size, orderby)
    table_  = self._table(table)
    columnd = { col.name:col for col in table_.columns }

    if isinstance(groupby, basestring):
      groupby = [ groupby ]
    if isinstance(aggregate, basestring):
      aggregate = [ aggregate ]

    session = self.sessionmaker()
    try:
      groupby_    = [ label(c, str2col(c, table_)) for c in groupby ]
      aggregate_  = [ label(a, str2col(a, table_)) for a in aggregate ]

      query = session.query(*(aggregate_ + groupby_))
      query = with_filters(query, table_, filters)
      query = query.group_by(*groupby_)
      query = with_orderby(query, table_, orderby)
      query = with_pagination(query, table_, page, page_size)
      result = result2dict(query.all())

      self.log.info("retrieved %d rows", len(result))
      return result

    finally:
      session.close()

    def __call__(self, user_ids, session):
        """
        Parameters:
            user_ids    : list of mediawiki user ids to restrict computation to
            session     : sqlalchemy session open on a mediawiki database

        Returns:
            {
                user id: 1 if they're a rolling new active editor, 0 otherwise
                for all cohort users, or all users that have edits in the time period
            }
        """
        number_of_edits = int(self.number_of_edits.data)
        rolling_days = int(self.rolling_days.data)
        end_date = self.end_date.data
        start_date = end_date - timedelta(days=rolling_days)

        newly_registered = session.query(Logging.log_user) \
            .filter(Logging.log_type == 'newusers') \
            .filter(Logging.log_action == 'create') \
            .filter(between(Logging.log_timestamp, start_date, end_date))

        filtered_new = self.filter(
            newly_registered, user_ids, column=Logging.log_user
        ).subquery()

        rev_user = label('user_id', Revision.rev_user)
        ar_user = label('user_id', Archive.ar_user)
        count = label('count', func.count())

        revisions = session.query(rev_user, count)\
            .filter(between(Revision.rev_timestamp, start_date, end_date))\
            .filter(Revision.rev_user.in_(filtered_new))\
            .group_by(Revision.rev_user)

        archived = session.query(ar_user, count)\
            .filter(between(Archive.ar_timestamp, start_date, end_date))\
            .filter(Archive.ar_user.in_(filtered_new))\
            .group_by(Archive.ar_user)

        bot_user_ids = session.query(MediawikiUserGroups.ug_user)\
            .filter(MediawikiUserGroups.ug_group == 'bot')\
            .subquery()

        new_edits = revisions.union_all(archived).subquery()
        new_edits_by_user = session.query(new_edits.c.user_id)\
            .filter(new_edits.c.user_id.notin_(bot_user_ids))\
            .group_by(new_edits.c.user_id)\
            .having(func.SUM(new_edits.c.count) >= number_of_edits)

        metric_results = {r[0]: {self.id : 1} for r in new_edits_by_user.all()}

        if user_ids is None:
            return metric_results
        else:
            return {
                uid: metric_results.get(uid, self.default_result)
                for uid in user_ids
            }

 def postgres_aggregates(self, resolution):
     if isinstance(resolution, basestring):
         try:
             resolution = float(resolution)
         except ValueError:
             resolution = self.resolution
     return [
         label('cell_x', func.floor(ST_X(Column('cell')) / resolution) * resolution),
         label('cell_y', func.floor(ST_Y(Column('cell')) / resolution) * resolution)]

    def __call__(self, user_ids, session):
        """
        Parameters:
            user_ids    : list of mediawiki user ids to find edit for
            session     : sqlalchemy session open on a mediawiki database

        Returns:
            dictionary from user ids to the number of edit found.
        """
        start_date = self.start_date.data
        end_date = self.end_date.data

        revisions = session\
            .query(
                label('user_id', Revision.rev_user),
                label('timestamp', Revision.rev_timestamp)
            )\
            .filter(Revision.rev_timestamp > start_date)\
            .filter(Revision.rev_timestamp <= end_date)

        archives = session\
            .query(
                label('user_id', Archive.ar_user),
                label('timestamp', Archive.ar_timestamp)
            )\
            .filter(Archive.ar_timestamp > start_date)\
            .filter(Archive.ar_timestamp <= end_date)

        if self.namespaces.data and len(self.namespaces.data) > 0:
            revisions = revisions.join(Page)\
                .filter(Page.page_namespace.in_(self.namespaces.data))
            archives = archives\
                .filter(Archive.ar_namespace.in_(self.namespaces.data))

        revisions = self.filter(revisions, user_ids, column=Revision.rev_user)
        archives = self.filter(archives, user_ids, column=Archive.ar_user)

        both = revisions
        if self.include_deleted.data:
            both = both.union_all(archives)
        both = both.subquery()

        query = session.query(both.c.user_id, func.count())\
            .group_by(both.c.user_id)

        query = self.apply_timeseries(query, column=both.c.timestamp)

        return self.results_by_user(
            user_ids,
            query,
            [(self.id, 1, 0)],
            date_index=2,
        )

    def build_query_to_report(self, query, aggregate_table, params):
        fk = Column(self.key, Integer)
        geom = Column(self.geometry_column, Geometry())
        join_table = Table(self.table, aggregate_table.metadata, fk, geom)
        if params == "key":
            query = query.column(label(self.key, aggregate_table.c.join_key))
        else:
            query = query.column(label("geometry", func.ST_AsGeoJSON(func.ST_Collect(geom))))

        return (
            query.select_from(join_table).where(aggregate_table.c.join_key == fk).group_by(aggregate_table.c.join_key)
        )

    def __call__(self, user_ids, session):
        """
        Parameters:
            user_ids    : list of mediawiki user ids to restrict computation to
            session     : sqlalchemy session open on a mediawiki database

        Returns:
            dictionary from user ids to: 1 if they're a rolling active editor, 0 if not
        """
        number_of_edits = int(self.number_of_edits.data)
        rolling_days = int(self.rolling_days.data)
        end_date = self.end_date.data
        start_date = end_date - timedelta(days=rolling_days)

        rev_user = label("user_id", Revision.rev_user)
        ar_user = label("user_id", Archive.ar_user)
        count = label("count", func.count())

        revisions = (
            session.query(rev_user, count)
            .filter(between(Revision.rev_timestamp, start_date, end_date))
            .group_by(Revision.rev_user)
        )
        revisions = self.filter(revisions, user_ids, column=Revision.rev_user)

        archived = (
            session.query(ar_user, count)
            .filter(between(Archive.ar_timestamp, start_date, end_date))
            .group_by(Archive.ar_user)
        )
        archived = self.filter(archived, user_ids, column=Archive.ar_user)

        bot_user_ids = (
            session.query(MediawikiUserGroups.ug_user).filter(MediawikiUserGroups.ug_group == "bot").subquery()
        )

        edits = revisions.union_all(archived).subquery()
        edits_by_user = (
            session.query(edits.c.user_id)
            .filter(edits.c.user_id.notin_(bot_user_ids))
            .group_by(edits.c.user_id)
            .having(func.SUM(edits.c.count) >= number_of_edits)
        )

        metric_results = {r[0]: {self.id: 1} for r in edits_by_user.all()}

        if user_ids is None:
            return metric_results
        else:
            return {uid: metric_results.get(uid, self.default_result) for uid in user_ids}

    def get_people_by_domain(self, session, limit=10):
        '''SELECT mailing_list_url, lower(domain_name) as domain,
                  count(lower(p.email_address)) as t
             FROM mailing_lists_people as ml, people as p
            WHERE lower(ml.email_address) = lower(p.email_address)
            GROUP BY mailing_list_url, domain
            ORDER BY t DESC, domain
            LIMIT %s;'''

        mailing_lists = int(self.get_num_of_mailing_lists(session)[0])
        limit = limit * mailing_lists

        mlp = aliased(db.MailingListsPeople)
        p = aliased(db.People)
        ret = session.query(mlp.mailing_list_url,
                            label('domain', func.lower(p.domain_name)),
                            func.count(func.lower(p.email_address)))\
                     .filter(func.lower(mlp.email_address) ==
                             func.lower(p.email_address))\
                     .group_by(mlp.mailing_list_url,
                               func.lower(p.domain_name))\
                     .order_by(func.count(func.lower(p.email_address)).desc(),
                               func.lower(p.domain_name))\
                     .limit(limit)
        return ret.all()

    def get_messages_by_domain(self, session, limit=10):
        '''SELECT m.mailing_list_url, lower(p.domain_name) as domain,
                  count(m.message_id) as num_messages
             FROM messages m,messages_people mp, people p
            WHERE m.message_ID = mp.message_ID
              AND lower(mp.email_address) = lower(p.email_address)
              AND mp.type_of_recipient = 'From'
            GROUP BY m.mailing_list_url, domain
            ORDER BY num_messages DESC, domain
            LIMIT %s;'''
        mailing_lists = int(self.get_num_of_mailing_lists(session)[0])
        limit = limit * mailing_lists

        m = aliased(db.Messages)
        mp = aliased(db.MessagesPeople)
        p = aliased(db.People)
        ret = session.query(m.mailing_list_url,
                            label('domain', func.lower(p.domain_name)),
                            func.count(m.message_id))\
                     .filter(m.message_id == mp.message_id)\
                     .filter(func.lower(mp.email_address) ==
                             func.lower(p.email_address))\
                     .filter(mp.type_of_recipient == 'From')\
                     .group_by(m.mailing_list_url,
                               func.lower(p.domain_name))\
                     .order_by(func.count(m.message_id).desc(),
                               func.lower(p.domain_name))\
                     .limit(limit)
        return ret.all()

def get_player_graph_data(server, granularity=15, start_date=None, end_date=None):
    end_date = end_date or datetime.utcnow()
    start_date = start_date or end_date - timedelta(days=7)

    result = db.session.query(
        label(
            'timestamp_group',
            func.round(
                (func.unix_timestamp(ServerStatus.timestamp) - time.timezone) / (granularity * 60)
            ),
        ),
        func.avg(ServerStatus.player_count)
    ).filter(
        ServerStatus.server == server,
        ServerStatus.timestamp >= start_date,
        ServerStatus.timestamp <= end_date
    ).group_by('timestamp_group').order_by(
        ServerStatus.timestamp
    ).all()

    points = []
    for chunk, count in result:
        points.append({
            'time': int(chunk * granularity * 60 * 1000),
            'player_count': int(count)
        })

    return {
        'start_time': int(calendar.timegm(start_date.timetuple()) * 1000),
        'end_time': int(calendar.timegm(end_date.timetuple()) * 1000),
        'points': points
    }

  def select(self, table, columns=None, filters={}, page=0, page_size=100, orderby=None):
    self.log.info((
        "table=%s, columns=%s, filters=%s, page=%s, page_size=%s, orderby=%s"
      ), table, columns, filters, page, page_size, orderby)
    table_  = self._table(table)
    columnd = { c.name:c for c in table_.columns }

    # get column objects corresponding to names
    if isinstance(columns, basestring):
      columns = [ columns ]

    if columns is None:
      columns_ = list(table_.columns)
    else:
      columns_ = [label(c, str2col(c, table_)) for c in columns]

    session = self.sessionmaker()
    try:
      query = session.query(*columns_)
      query = with_filters(query, table_, filters)
      query = with_orderby(query, table_, orderby)
      query = with_pagination(query, table_, page, page_size)
      result = result2dict(query.all())
      self.log.info("retrieved %d rows", len(result))
      return result
    finally:
      session.close()

    def column(self, field, dimension = None):
        """Return a table column for `field` which can be either :class:`cubes.Attribute` or a string.
        
        Possible column names:
        * ``field`` for fact field or flat dimension
        * ``field.locale`` for localized fact field or flat dimension
        * ``dimension.field`` for multi-level dimension field
        * ``dimension.field.locale`` for localized multi-level dimension field
        """

        # FIXME: should use: field.full_name(dimension, self.locale)
        # if there is no localization for field, use default name/first locale
        locale_suffix = ""

        if isinstance(field, cubes.model.Attribute) and field.locales:
            locale = self.locale if self.locale in field.locales else field.locales[0]
            locale_suffix = "." + locale

        if dimension:
            # FIXME: temporary flat dimension hack, not sure about impact of this to other parts of the
            # framework
            if not dimension.is_flat or dimension.has_details:
                logical_name = dimension.name + '.' + str(field)
            else:
                logical_name = str(field)
        else:
            logical_name = field

        self.logger.debug("getting column %s(%s) loc: %s - %s" % (field, type(field), self.locale, locale_suffix))

        localized_name = logical_name + locale_suffix
        column = self.view.c[localized_name]
        return expression.label(logical_name, column)

def SelectSingeMenuPrivilege(strUserID, MidList):
    """
    @note 查询列表里菜单ID的权限
    :param strUserID:
    :param MidList:
    :return: 返回菜单权限列表
    """
    project_dic = CommonSession.SelectProject('ProjectDic')
    menu_list = []
    with GetSession() as db_ses:
        privilege = db_ses.query(tables.MenuPrivilege.mid, tables.Menu.name, tables.Menu.url, tables.Menu.preid,
                                 expression.label('privileges', func.group_concat(tables.MenuPrivilege.pid, ";",
                                                                                  tables.MenuPrivilege.r_priv, ";",
                                                                                  tables.MenuPrivilege.w_priv))).join(
            tables.Menu, tables.MenuPrivilege.mid == tables.Menu.mid).filter(tables.MenuPrivilege.uid == strUserID,
                                                                             tables.MenuPrivilege.mid.in_(
                                                                                 MidList)).group_by(
            tables.MenuPrivilege.mid).all()
        for menu in privilege:
            priv_list = []
            for prjs in str(menu[4]).split(','):

                priv = prjs.split(';')
                prj_dic = {}
                if priv[0] in project_dic.keys():
                    prj_dic[project_dic[priv[0]]] = {'pid': priv[0], 'r_priv': priv[1], 'w_priv': priv[2]}
                    priv_list.append(prj_dic)

            menu_dic = {'menu_id': menu[0], 'menu_name': menu[1], 'menu_url': menu[2], 'menu_preid': menu[3],
                        'menu_pri': priv_list}
            menu_list.append(menu_dic)
    return menu_list

    def build_query_to_report(self, query, aggregate_table, params):
        assert params in self._known_units
        res = params

        truncated_time = func.date_trunc(res, aggregate_table.c.time_step)
        return (query
                .column(label("time_slice", func.extract("epoch", truncated_time)))
                .group_by(truncated_time))

def SelectMenuProjectPrivilege(strUserId, strMenuID='None'):
    """
    @note 查询用户菜单权限
    :param strUserId:
    :param strMenuID: None 返回此用户所有菜单权限
                       不为None 返回此用户某个菜单ID的菜单权限
    :return:
    """

    project_dic = CommonSession.SelectProject('ProjectDic')

    menu_list = []
    with GetSession() as db_ses:
        if strMenuID == 'None':
            privilege = db_ses.query(tables.MenuPrivilege.mid, tables.Menu.name, tables.Menu.url, tables.Menu.preid,
                                     expression.label('privileges', func.group_concat(tables.MenuPrivilege.pid, ";",
                                                                                      tables.MenuPrivilege.r_priv, ";",
                                                                                      tables.MenuPrivilege.w_priv))).join(
                tables.Menu, tables.MenuPrivilege.mid == tables.Menu.mid).filter(
                tables.MenuPrivilege.uid == strUserId).group_by(tables.MenuPrivilege.mid).all()

        else:
            privilege = db_ses.query(tables.MenuPrivilege.mid, tables.Menu.name, tables.Menu.url, tables.Menu.preid,
                                     expression.label('privileges', func.group_concat(tables.MenuPrivilege.pid, ";",
                                                                                      tables.MenuPrivilege.r_priv, ";",
                                                                                      tables.MenuPrivilege.w_priv))).join(
                tables.Menu, tables.MenuPrivilege.mid == tables.Menu.mid).filter(
                tables.MenuPrivilege.uid == strUserId, tables.MenuPrivilege.mid == strMenuID).group_by(
                tables.MenuPrivilege.mid).all()
        for menu in privilege:
            priv_list = []
            for prjs in str(menu[4]).split(','):

                priv = prjs.split(';')
                prj_dic = {}
                if priv[0] in project_dic.keys():
                    prj_dic[project_dic[priv[0]]] = {'pid': priv[0], 'r_priv': priv[1], 'w_priv': priv[2]}
                    priv_list.append(prj_dic)

            menu_dic = {'menu_id': menu[0], 'menu_name': menu[1], 'menu_url': menu[2], 'menu_preid': menu[3],
                        'menu_pri': priv_list}
            menu_list.append(menu_dic)
    return menu_list

def get(run_id, query, with_stats=True):
    """Return a list of genotypes in a vcf conforming to the given query, as
    well as a dict of stats calculated on them.

    If a truth_vcf is associated with this VCF, stats include true/false,
    positive/negative stats, as well as precision, recall, and f1score. Stats
    also include the number of records, and the number of records once filters
    are applied.

    A query is a dictionary which specifies the range, filters, limit, offset
    and ordering which should be applied against genotypes before genotypes and
    stats are returned.

    It has structure:

    {range: {contig: "X", start: 0, end: 250000000},
     filters: [{columnName: 'info:DP', filterValue: '50', type: '<'}, ...],
     sortBy: [{columnName: 'contig', order: 'asc'},
              {columnName: 'position', order: 'asc'}, ...],
     page: 10,
     limit: 250
    }
    """
    query = _annotate_query_with_types(query, spec(run_id))
    compare_to_run_id = query.get('compareToVcfId')
    with tables(db.engine, 'genotypes') as (con, g):
        if compare_to_run_id:
            # We consider a genotype validated if a truth genotype exists at its
            # location (contig/position) with the same ref/alts.  This isn't
            # entirely accurate: for example, it handles SVs very poorly.
            gt = g.alias()
            joined_q = outerjoin(g, gt, and_(
                gt.c.vcf_id == compare_to_run_id,
                g.c.contig == gt.c.contig,
                g.c.position == gt.c.position,
                g.c.reference == gt.c.reference,
                g.c.alternates == gt.c.alternates,
                g.c.sample_name == gt.c.sample_name))
            valid_column = label('tag:true-positive', gt.c.contig != None)
            q = (select(g.c + [valid_column])
                 .select_from(joined_q)
                 .where(g.c.vcf_id == run_id))
        else:
            q = select(g.c).where(g.c.vcf_id == run_id)

        q = _add_range(q, g, query.get('range'))
        q = _add_filters(q, g, query.get('filters'))
        q = _add_orderings(q, g, query.get('sortBy'))
        q = _add_paging(q, g, query.get('limit'), query.get('page'))

        q = _add_ordering(q, g, 'String', 'contig', 'asc')
        q = _add_ordering(q, g, 'Integer', 'position', 'asc')
        genotypes = [dict(g) for g in con.execute(q).fetchall()]
    stats = calculate_stats(run_id, compare_to_run_id, query) if with_stats else {}
    return {'records': genotypes, 'stats': stats}

    def build_query_to_populate(self, query, full_table, aggregate_table):
        insert_columns = [aggregate_table.c.isp]
        ip_range = Column("ip_range", INT8RANGE)
        isp_name = Column("label", String)
        join_table = Table(self.maxmind_table, full_table.metadata, ip_range, isp_name, keep_existing = True)
        isp_label = label('maxmind_isp', self._sql_rewrite(isp_name))
        select_query = (query.select_from(join_table)
                .where(ip_range.contains(full_table.c.client_ip))
                .column(isp_label)
                .group_by('maxmind_isp'))

        return insert_columns, select_query

    def get_messages_by_people(self, session, limit=10):
        '''SELECT m.mailing_list_url, lower(mp.email_address) as email,
                  count(m.message_ID) as t
             FROM messages m, messages_people mp
            WHERE m.message_ID = mp.message_ID
              AND mp.type_of_recipient = 'From'
            GROUP BY m.mailing_list_url, email
            ORDER BY t desc, email limit %s;'''

        m = aliased(db.Messages)
        mp = aliased(db.MessagesPeople)
        ret = session.query(m.mailing_list_url,
                            label('email', func.lower(mp.email_address)),
                            label('t', func.count(m.message_id)))\
                     .filter(m.message_id == mp.message_id)\
                     .filter(mp.type_of_recipient == 'From')\
                     .group_by(m.mailing_list_url,
                               func.lower(mp.email_address))\
                     .order_by(func.count(m.message_id).desc(),
                               func.lower(mp.email_address))\
                     .limit(limit)
        return ret.all()

    def _select_column(self, attribute, locale = None):
        """get select column"""
        
        if locale:
            localized_alias = attribute.alias + "." + locale
        else:
            localized_alias = attribute.alias

        if self.dimension_table_prefix:
            prefix = self.dimension_table_prefix
        else:
            prefix = ""
        self.logger.debug("looking for mapping %s (%s)" % (localized_alias, attribute.alias))

        if self.cube.mappings and localized_alias in self.cube.mappings:
            mapping = self.cube.mappings[localized_alias]
            original_mapping = mapping
            self.logger.debug("  is in mappings: %s" % mapping)
        elif self.cube.mappings and attribute.alias in self.cube.mappings:
            mapping = self.cube.mappings[attribute.alias]
            original_mapping = mapping
            self.logger.debug("  not in mappings, using default trans: %s" % mapping)
        else:
            original_mapping = None
            if attribute.dimension:
                mapping = prefix + attribute.alias
            else:
                mapping = attribute.alias

            # FIXME: make this work
            if locale:
                mapping = mapping + "_" + locale
                
            self.logger.debug("  defaulting to: %s" % mapping)

        (table_name, field_name) = self.split_field(mapping)
        if not table_name:
            table_name = self.fact_name
            
        table = self.table(table_name)

        try:
            column = table.c[field_name]
        except KeyError:
            raise model.ModelError("Mapped column '%s' does not exist (as %s.%s)" \
                                        % (localized_alias, table_name, field_name) )
        
        self.logger.debug("adding column %s as %s" % (column, localized_alias))
        # self.mappings[localized_alias] = column
        return expression.label(localized_alias, column)

def generate_show():
    subquery = (db.session.query(LikeModel.media,
                                 func.count(1).label('count'))
                .group_by(LikeModel.media).subquery())
    now = int(time.time() / 7200)
    order = expression.label('hacker', (subquery.c.count + 1.0) / (now - ShowModel.hour_tagged + 2.0) / (now - ShowModel.hour_tagged + 2.0))
    medias =\
        (db.session.query(ShowModel)
         .filter(ShowModel.showable == 0)
         .outerjoin(subquery, ShowModel.mid == subquery.c.media)
         .filter(ShowModel.mid != None)     # NOQA
         .order_by(order.desc())
         .order_by(ShowModel.date_tagged.desc())
         .order_by(ShowModel.date_created.desc())
         .all())
    return [x.mid for x in medias]

def get_slow_pages(path=None):
    l = label("average", func.avg(ViewLog.load_time))
    c = label("count", func.count(ViewLog.id))

    return (
        DBSession.query(
            ViewLog.path,
            l,
            c,
            label("cumulative_time", func.sum(ViewLog.load_time)),
            label("stddev", func.stddev_pop(ViewLog.load_time)),
            label("maximum", func.max(ViewLog.load_time)),
            label("minimum", func.min(ViewLog.load_time)),
        )
        .filter(ViewLog.path == path if path != None else True)
        .having(c > 2)
        .group_by(ViewLog.path)
        .order_by(l.desc())
    )