<commit_before>import hashlib
from sqlagg import SumWhen
import sqlalchemy
from django.conf import settings
from sqlalchemy.exc import IntegrityError
from corehq.db import Session
from corehq.apps.reports.sqlreport import DatabaseColumn
from dimagi.utils.decorators.memoized import memoized
from fluff.util import get_column_type

metadata = sqlalchemy.MetaData()


class IndicatorSqlAdapter(object):

    def __init__(self, engine, config):
        self.engine = engine
        self.config = config

    @memoized
    def get_table(self):
        return get_indicator_table(self.config)

    def rebuild_table(self):
        rebuild_table(self.engine, self.get_table())

    def drop_table(self):
        with self.engine.begin() as connection:
            self.get_table().drop(connection, checkfirst=True)

    def save(self, doc):
        indicator_rows = self.config.get_all_values(doc)
        if indicator_rows:
            table = self.get_table()
            for indicator_row in indicator_rows:
                with self.engine.begin() as connection:
                    # delete all existing rows for this doc to ensure we aren't left with stale data
                    delete = table.delete(table.c.doc_id == doc['_id'])
                    connection.execute(delete)
                    all_values = {i.column.id: i.value for i in indicator_row}
                    insert = table.insert().values(**all_values)
                    try:
                        connection.execute(insert)
                    except IntegrityError:
                        # Someone beat us to it. Concurrent inserts can happen
                        # when a doc is processed by the celery rebuild task
                        # at the same time as the pillow.
                        pass

    def delete(self, doc):
        table = self.get_table()
        with self.engine.begin() as connection:
            delete = table.delete(table.c.doc_id == doc['_id'])
            connection.execute(delete)


def get_indicator_table(indicator_config, custom_metadata=None):
    sql_columns = [column_to_sql(col) for col in indicator_config.get_columns()]
    table_name = get_table_name(indicator_config.domain, indicator_config.table_id)
    # todo: needed to add extend_existing=True to support multiple calls to this function for the same table.
    # is that valid?
    return sqlalchemy.Table(
        table_name,
        custom_metadata or metadata,
        extend_existing=True,
        *sql_columns
    )


def column_to_sql(column):
    return sqlalchemy.Column(
        column.id,
        get_column_type(column.datatype),
        nullable=column.is_nullable,
        primary_key=column.is_primary_key,
    )


def get_engine():
    return sqlalchemy.create_engine(settings.SQL_REPORTING_DATABASE_URL)


def rebuild_table(engine, table):
    with engine.begin() as connection:
        table.drop(connection, checkfirst=True)
        table.create(connection)
    engine.dispose()


def get_column_name(path):
    parts = path.split("/")

    def _hash(parts):
        front = "/".join(parts[:-1])
        end = parts[-1]
        return hashlib.sha1('{}_{}'.format(hashlib.sha1(front).hexdigest(), end)).hexdigest()[:8]

    return "_".join(parts + [_hash(parts)])


def get_table_name(domain, table_id):
    def _hash(domain, table_id):
        return hashlib.sha1('{}_{}'.format(hashlib.sha1(domain).hexdigest(), table_id)).hexdigest()[:8]

    return 'config_report_{0}_{1}_{2}'.format(domain, table_id, _hash(domain, table_id))


def get_expanded_columns(data_source_configuration, column_config, column_warnings):
    '''
    Given a ReportColumn, return a list of DatabaseColumn objects. Each DatabaseColumn
    is configured to show the number of occurences of one of the values present for
    the ReportColumn's field.

    This function also adds warnings to the column_warnings parameter.

    :param data_source_configuration:
    :param column_config:
    :param column_warnings:
    :return:
    '''
    MAXIMUM_EXPANSION = 10

    session = Session()
    try:
        connection = session.connection()
        table = get_indicator_table(data_source_configuration)
        if not table.exists(bind=connection):
            return []
        column = table.c[column_config.field]

        query = sqlalchemy.select([column], limit=MAXIMUM_EXPANSION + 1).distinct()
        result = connection.execute(query).fetchall()
        distinct_values = [x[0] for x in result]
        if len(distinct_values) > MAXIMUM_EXPANSION:
            distinct_values = distinct_values[:MAXIMUM_EXPANSION]
            column_warnings.append(
                'The "{}" column had too many values to expand! '
                'Expansion limited to {} distinct values.'.format(
                    column_config.display, MAXIMUM_EXPANSION
                )
            )
    except:
        session.rollback()
        raise
    finally:
        session.close()

    columns = []
    for val in distinct_values:
        columns.append(DatabaseColumn(
            u"{}-{}".format(column_config.display, val),
            SumWhen(
                column_config.field,
                whens={val: 1},
                else_=0,
                alias=u"{}-{}".format(column_config.report_column_id, val),
            ),
            sortable=False,
            data_slug=u"{}-{}".format(column_config.report_column_id, val),
            format_fn=column_config.get_format_fn(),
            help_text=column_config.description
        ))
    return columns<commit_msg>break get_expanded_columns into smaller chunks.<commit_after>import hashlib
from sqlagg import SumWhen
import sqlalchemy
from django.conf import settings
from sqlalchemy.exc import IntegrityError
from corehq.db import Session
from corehq.apps.reports.sqlreport import DatabaseColumn
from dimagi.utils.decorators.memoized import memoized
from fluff.util import get_column_type

metadata = sqlalchemy.MetaData()


class IndicatorSqlAdapter(object):

    def __init__(self, engine, config):
        self.engine = engine
        self.config = config

    @memoized
    def get_table(self):
        return get_indicator_table(self.config)

    def rebuild_table(self):
        rebuild_table(self.engine, self.get_table())

    def drop_table(self):
        with self.engine.begin() as connection:
            self.get_table().drop(connection, checkfirst=True)

    def save(self, doc):
        indicator_rows = self.config.get_all_values(doc)
        if indicator_rows:
            table = self.get_table()
            for indicator_row in indicator_rows:
                with self.engine.begin() as connection:
                    # delete all existing rows for this doc to ensure we aren't left with stale data
                    delete = table.delete(table.c.doc_id == doc['_id'])
                    connection.execute(delete)
                    all_values = {i.column.id: i.value for i in indicator_row}
                    insert = table.insert().values(**all_values)
                    try:
                        connection.execute(insert)
                    except IntegrityError:
                        # Someone beat us to it. Concurrent inserts can happen
                        # when a doc is processed by the celery rebuild task
                        # at the same time as the pillow.
                        pass

    def delete(self, doc):
        table = self.get_table()
        with self.engine.begin() as connection:
            delete = table.delete(table.c.doc_id == doc['_id'])
            connection.execute(delete)


def get_indicator_table(indicator_config, custom_metadata=None):
    sql_columns = [column_to_sql(col) for col in indicator_config.get_columns()]
    table_name = get_table_name(indicator_config.domain, indicator_config.table_id)
    # todo: needed to add extend_existing=True to support multiple calls to this function for the same table.
    # is that valid?
    return sqlalchemy.Table(
        table_name,
        custom_metadata or metadata,
        extend_existing=True,
        *sql_columns
    )


def column_to_sql(column):
    return sqlalchemy.Column(
        column.id,
        get_column_type(column.datatype),
        nullable=column.is_nullable,
        primary_key=column.is_primary_key,
    )


def get_engine():
    return sqlalchemy.create_engine(settings.SQL_REPORTING_DATABASE_URL)


def rebuild_table(engine, table):
    with engine.begin() as connection:
        table.drop(connection, checkfirst=True)
        table.create(connection)
    engine.dispose()


def get_column_name(path):
    parts = path.split("/")

    def _hash(parts):
        front = "/".join(parts[:-1])
        end = parts[-1]
        return hashlib.sha1('{}_{}'.format(hashlib.sha1(front).hexdigest(), end)).hexdigest()[:8]

    return "_".join(parts + [_hash(parts)])


def get_table_name(domain, table_id):
    def _hash(domain, table_id):
        return hashlib.sha1('{}_{}'.format(hashlib.sha1(domain).hexdigest(), table_id)).hexdigest()[:8]

    return 'config_report_{0}_{1}_{2}'.format(domain, table_id, _hash(domain, table_id))


def get_expanded_columns(data_source_configuration, column_config, column_warnings):
    """
    Given a ReportColumn, return a list of DatabaseColumn objects. Each DatabaseColumn
    is configured to show the number of occurrences of one of the values present for
    the ReportColumn's field.

    This function also adds warnings to the column_warnings parameter.

    :param data_source_configuration:
    :param column_config:
    :param column_warnings:
    :return:
    """
    MAXIMUM_EXPANSION = 10
    vals, over_expansion_limit = _get_distinct_values(
        data_source_configuration, column_config, MAXIMUM_EXPANSION
    )
    if over_expansion_limit:
        column_warnings.append(
            'The "{}" column had too many values to expand! '
            'Expansion limited to {} distinct values.'.format(
                column_config.display, MAXIMUM_EXPANSION
            )
        )
    return _expand_column(column_config, vals)


def _get_distinct_values(data_source_configuration, column_config, expansion_limit=10):
    """
    Return a tuple. The first item is a list of distinct values in the given
    ReportColumn no longer than expansion_limit. The second is a boolean which
    is True if the number of distinct values in the column is greater than the
    limit.

    :param data_source_configuration:
    :param column_config:
    :param expansion_limit:
    :return:
    """
    too_many_values = False

    session = Session()
    try:
        connection = session.connection()
        table = get_indicator_table(data_source_configuration)
        if not table.exists(bind=connection):
            return []
        column = table.c[column_config.field]

        query = sqlalchemy.select([column], limit=expansion_limit + 1).distinct()
        result = connection.execute(query).fetchall()
        distinct_values = [x[0] for x in result]
        if len(distinct_values) > expansion_limit:
            distinct_values = distinct_values[:expansion_limit]
            too_many_values = True
    except:
        session.rollback()
        raise
    finally:
        session.close()

    return distinct_values, too_many_values


def _expand_column(report_column, distinct_values):
    """
    Given a ReportColumn, return a list of DatabaseColumn objects. Each column
    is configured to show the number of occurrences of one of the given distinct_values.

    :param report_column:
    :param distinct_values:
    :return:
    """
    columns = []
    for val in distinct_values:
        columns.append(DatabaseColumn(
            u"{}-{}".format(report_column.display, val),
            SumWhen(
                report_column.field,
                whens={val: 1},
                else_=0,
                alias=u"{}-{}".format(report_column.report_column_id, val),
            ),
            sortable=False,
            data_slug=u"{}-{}".format(report_column.report_column_id, val),
            format_fn=report_column.get_format_fn(),
            help_text=report_column.description
        ))
    return columns
{'fraction_non_alphanumeric': 0.09012558483132234, 'fraction_numerical': 0.002954937207584339, 'mean_word_length': 2.5552961774146485, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3077916', 'n_tokens_mistral': 3500, 'n_tokens_neox': 3366, 'n_words': 839}